wenfangwen_repository/work_03/langgraph_practice.py

from langgraph.graph import START, END, StateGraph, MessagesState
from typing import TypedDict, Annotated
from operator import add
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os


load_dotenv(override=True)
deepseek_base_url = os.getenv("DEEPSEEK_BASE_URL")
deepseek_base_key = os.getenv("DEEPSEEK_BASE_KEY")
deepseek_base_name = os.getenv("DEEPSEEK_BASE_NAME")

print("1. 三点===============================================================================================================================================================================================================")
class CounterState1(TypedDict):
    count: int

def incr_node_1(state: CounterState1)-> dict:
    count = state["count"] + 1
    return {"count": count}

def incr_node_2(state: CounterState1)-> dict:
    count = state["count"] * 2
    return {"count": count}

build = StateGraph(CounterState1)
build.add_node("incr_node_1", incr_node_1)
build.add_node("incr_node_2", incr_node_2)

build.add_edge(START, "incr_node_1")
build.add_edge("incr_node_1", "incr_node_2")
build.add_edge("incr_node_2", END)

graph = build.compile()
result = graph.invoke({"count": 1})
print(result)


print("2. 条件边===============================================================================================================================================================================================================")
class CounterState2(TypedDict):
    count: int

def sub_node_1(state: CounterState2) -> dict:
    count = state["count"] - 1
    return {"count": count}

def sub_node_2(state: CounterState2) -> dict:
    count = state["count"] - 2
    return {"count": count}

def condition_edge(state: CounterState2) -> str:
    if state["count"] > 5:
        return "sub_node_1"
    return "sub_node_2"

build = StateGraph(CounterState2)

build.add_node("sub_node_1", sub_node_1).add_node("sub_node_2", sub_node_2)

build.add_edge(START, "sub_node_1")
build.add_conditional_edges(
    "sub_node_1",
    condition_edge,
    {"sub_node_1": "sub_node_1", "sub_node_2": "sub_node_2"}
)
build.add_edge("sub_node_2", END)

graph = build.compile()
result = graph.invoke({"count": 20})
print(result)

print("3. 覆盖===============================================================================================================================================================================================================")
class OverwriteState(TypedDict):
    arr:  list[str]

def node_1(state: OverwriteState)->dict:
    return {"arr":["a", "b"]}

def node_2(state: OverwriteState)->dict:
    return {"arr":["c", "d"]}

build = StateGraph(OverwriteState)
build.add_node("node_1", node_1).add_node("node_2", node_2)
build.add_edge(START, "node_1")
build.add_edge("node_1", "node_2")
build.add_edge("node_2", END)

graph = build.compile()
result = graph.invoke({"arr":["e", "f"]})
print(result)

print("4. 追加===============================================================================================================================================================================================================")
class AppendState(TypedDict):
    arr: Annotated[list[str], add]

def node_1(state: AppendState)->dict:
    return {"arr":["aa", "bb"]}

def node_2(state: AppendState)->dict:
    return {"arr":["cc", "dd"]}

build = StateGraph(AppendState)
build.add_node("node_1", node_1).add_node("node_2", node_2)
build.add_edge(START, "node_1")
build.add_edge("node_1", "node_2")
build.add_edge("node_2", END)

graph = build.compile()
result = graph.invoke({"arr":["ee", "ff"]})
print(result)

print("5. API===============================================================================================================================================================================================================")
# llm = ChatOpenAI(
#     base_url= deepseek_base_url,
#     api_key= deepseek_base_key,
#     model= deepseek_base_name
# )

# def chat_llm(state: MessagesState)->dict:
#     response = llm.invoke("你是谁")
#     return {"messages": [response]}

# build = StateGraph(MessagesState)
# build.add_node("chat", chat_llm)
# build.add_edge(START, "chat")
# build.add_edge("chat", END)

# graph = build.compile()
# result = graph.invoke({"messages": ["多学多问"]})
# print(result)

print("6. 自定义===============================================================================================================================================================================================================")

class ReplaceableList(list):
    def __init__(self, *args, replace: bool = False, **kwargs):
        super().__init__(*args, **kwargs)
        self.replace = replace

def smart_merge(current: list, incom: list)->list:
    if isinstance(incom, ReplaceableList) and getattr(incom, "replace", False):
        return list(incom)
    return current + incom

class CostomState(TypedDict):
    arr: Annotated[list[str], smart_merge]

def append(state: CostomState)->dict:
    return {"arr": ["b", "c"]}
                 
def overwrite(state: CostomState)->dict:
    new_list = ReplaceableList(["reset_data"])
    new_list.replace = True
    return {"arr": new_list}

build = StateGraph(CostomState)
build.add_node("append", append)
build.add_node("overwrite", overwrite)
build.add_edge(START, "overwrite")
build.add_edge("overwrite", "append")
build.add_edge("append", END)

graph = build.compile()
result = graph.invoke({"arr": ["a"]})
print(result)


print("7. 子图===============================================================================================================================================================================================================")

class DrinkState(TypedDict):
    logs: Annotated[list[str], add]

class MakeState(TypedDict):
    sub_logs: Annotated[list[str], add]

def add_coffee(state: MakeState) -> dict:
    return {"sub_logs": ["添加咖啡"]}

def add_sugar(state: MakeState) -> dict:
    return {"sub_logs": ["添加粮"]}

def do_make_coffee(state: MakeState) -> dict:
    return {"sub_logs": ["咖啡制作中..."]}

makeBuild = StateGraph(MakeState)
makeBuild.add_node("add_coffee", add_coffee)
makeBuild.add_node("add_sugar", add_sugar)
makeBuild.add_node("do_make_coffee", do_make_coffee)
makeBuild.add_edge(START, "add_coffee")
makeBuild.add_edge("add_coffee", "add_sugar")
makeBuild.add_edge("add_sugar", "do_make_coffee")
makeBuild.add_edge("do_make_coffee", END)
makeGraph = makeBuild.compile()

def order_coffee(state: DrinkState) -> dict:
    return {"logs": ["点咖啡"]}

def make_coffee(state: DrinkState) -> dict:
    result = makeGraph.invoke({"sub_logs": ["开始制作咖啡"]})
    return {"logs": result["sub_logs"]}

def drink_coffee(state: DrinkState) -> dict:
    return {"logs": ["喝咖啡"]}

build = StateGraph(DrinkState)
build.add_node("order_coffee", order_coffee)
build.add_node("make_coffee", make_coffee)
build.add_node("drink_coffee", drink_coffee)
build.add_edge(START, "order_coffee")
build.add_edge("order_coffee", "make_coffee")
build.add_edge("make_coffee", "drink_coffee")
build.add_edge("drink_coffee", END)

graph = build.compile()
result = graph.invoke({"logs": ["喝咖啡"]})
print(result)


print("8. Send===============================================================================================================================================================================================================")
"""
图的拓扑结构：

            START
              │
              ▼
        [parse_query]            ← 节点：解析用户输入，提取品牌列表
              │
              ▼
        route_to_research        ← 条件边路由函数：返回 N 个 Send
              │
              ├─ Send("research_brand", {brand: "苹果"})
              ├─ Send("research_brand", {brand: "华为"})    ← 三个并行
              └─ Send("research_brand", {brand: "小米"})
              │
              ▼  (并行执行)
        [research_brand] × 3     ← 每个品牌独立研究
              │
              ▼  (全部完成后汇总)
        [summarize]              ← 节点：汇总所有报告，输出对比
              │
              ▼
             END
"""
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send


# ══════════════════════════════════════════════════════════════
#  State 定义
# ══════════════════════════════════════════════════════════════
class CompareState(TypedDict):
    """手机对比任务的 State"""
    query: str                                # 用户的原始问题
    brands: list[str]                         # 从 query 中解析出的品牌列表
    reports: Annotated[list[str], add]        # 每个品牌的研究报告（add reducer 自动合并）
    comparison: str                           # 最终的对比总结


# ══════════════════════════════════════════════════════════════
#  节点函数
# ══════════════════════════════════════════════════════════════
def parse_query(state: CompareState) -> dict:
    """
    节点 ①：解析用户输入，提取出要对比的品牌列表。

    实际场景中这里会用 LLM 做实体提取，这里直接硬编码模拟。
    """
    query = state["query"]
    print(f"[parse] 收到问题: {query}")

    # 模拟 LLM 解析：从自然语言中提取品牌名称
    brands = ["苹果", "华为", "小米"]
    print(f"[parse] 解析出 {len(brands)} 个品牌: {brands}")

    return {"brands": brands}


def research_brand(state: CompareState) -> dict:
    """
    节点 ②：研究单个品牌（被 Send 并行调用的目标节点）。

    它不知道总共有几个品牌，也不关心其他品牌在做什么——
    只专注于自己拿到的这一个。

    实际场景中这里会调搜索引擎 API 或 LLM，这里用数据模拟。
    """
    brand = state["brands"][0]  # Send 保证这里只有一个品牌

    # 模拟：查询该品牌的旗舰机信息
    phone_db = {
        "苹果": "iPhone 16 Pro Max | A18 Pro 芯片 | 6.9吋 OLED | 4K 120fps 视频 | iOS 18",
        "华为": "Mate 70 Pro+ | 麒麟 9100 | 6.8吋 OLED | 物理可变光圈 | 鸿蒙 NEXT",
        "小米": "小米 15 Ultra | 骁龙 8 Gen4 | 6.73吋 AMOLED | 徕卡光学 | 澎湃 OS 2.0",
    }

    info = phone_db.get(brand, f"{brand}旗舰机信息暂缺")
    report = f"【{brand}】{info}"

    print(f"  [research] 并行查询 [{brand}] -> {info}")
    return {"reports": [report]}  # add reducer 自动拼到总 reports 里


def summarize(state: CompareState) -> dict:
    """
    节点 ③：所有品牌都研究完了，汇总生成对比结论。

    实际场景中这里会把所有 reports 喂给 LLM，让它生成对比分析。
    """
    print(f"\n[summarize] 开始汇总 {len(state['reports'])} 份报告...")

    # 模拟 LLM 生成的对比总结
    comparison = (
        "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n"
        "         >> 旗舰手机对比总结\n"
        "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n\n"
    )
    for report in state["reports"]:
        comparison += f"  {report}\n"

    comparison += (
        "\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n"
        "[总结] 三款旗舰各有千秋 --\n"
        "  苹果 iPhone 16 Pro Max：视频拍摄王者，生态闭环体验最佳\n"
        "  华为 Mate 70 Pro+：影像系统物理可变光圈独树一帜\n"
        "  小米 15 Ultra：徕卡光学加持，性价比旗舰首选\n"
        "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n"
    )

    return {"comparison": comparison}


# ══════════════════════════════════════════════════════════════
#  路由函数 —— 核心！返回 Send 列表而不是 dict
# ══════════════════════════════════════════════════════════════
def route_to_research(state: CompareState):
    """
    ★ Send API 的核心：条件边路由函数 ★

    关键规则：
    - 这个函数只能放在 add_conditional_edges 里，不能放进 add_node
    - 普通节点返回 dict（更新 state）
    - 路由函数返回 Send 列表（创建并行任务）或字符串（END / 节点名）

    思想：几个品牌 → 发几个 Send → 启动几次 research_brand

            return [
                Send("research_brand", {"brands": ["苹果"]}),   ─┐
                Send("research_brand", {"brands": ["华为"]}),   ─┤ 全部并行执行
                Send("research_brand", {"brands": ["小米"]}),   ─┘
            ]

    每个 Send 的两个参数：
      参数 1: 目标节点名 —— "派给谁做"
      参数 2: 此分支专属的 state —— "这个任务要什么数据"
    """
    if not state["brands"]:
        return END

    return [
        Send("research_brand", {"brands": [brand]})
        for brand in state["brands"]
    ]


# ══════════════════════════════════════════════════════════════
#  构图
# ══════════════════════════════════════════════════════════════
builder = StateGraph(CompareState)

builder.add_node("parse_query", parse_query)
builder.add_node("research_brand", research_brand)
builder.add_node("summarize", summarize)

# 边
builder.add_edge(START, "parse_query")
builder.add_conditional_edges("parse_query", route_to_research)  # 路由 → 动态 fan-out
builder.add_edge("research_brand", "summarize")                  # 所有并行结果汇聚到 summarize
builder.add_edge("summarize", END)

graph = builder.compile()


# ══════════════════════════════════════════════════════════════
#  运行
# ══════════════════════════════════════════════════════════════

result = graph.invoke({
    "query": "帮我对比一下苹果、华为和小米的旗舰机型",
    "brands": [],
    "reports": [],
    "comparison": "",
})
print("======")
print(result["brands"])
print(result["comparison"])