🔀 条件边与 Send

def add_conditional_edges(
    self,
    source: str,                          # 起点节点
    path: Callable[..., str | list[str]], # 路由函数：返回下一步节点名
    path_map: dict | list | None = None,  # 可选：把返回值映射到节点名
) -> Self:
    """添加条件边。执行完 source 后，由 path 决定去哪。

    Args:
        source: 起点节点。
        path: 决定下一步的函数。返回节点名、节点名列表，或 'END'。
        path_map: 可选映射。若提供，path 返回的值会经 path_map 翻译成节点名。

    注意：path 返回 'END' 时图停止执行。
    """

# pip install langgraph
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    query: str
    category: str
    response: str

def classify(state: State) -> dict:
    """节点1：给查询分类（这里假装是 LLM 分类）。"""
    q = state["query"]
    if "天气" in q:
        cat = "weather"
    elif "计算" in q or "数学" in q:
        cat = "math"
    else:
        cat = "general"
    return {"category": cat}

def weather_agent(state: State) -> dict:
    return {"response": f"天气预报：{state['query']}"}

def math_agent(state: State) -> dict:
    return {"response": f"数学计算：{state['query']}"}

def general_agent(state: State) -> dict:
    return {"response": f"通用回答：{state['query']}"}

def route(state: State) -> str:
    """条件边的 path 函数：根据 category 路由到不同专家。"""
    return state["category"]   # 返回值 = 目标节点名

# 构建图
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("weather", weather_agent)
graph.add_node("math", math_agent)
graph.add_node("general", general_agent)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route)   # ← 条件边！
graph.add_edge("weather", END)
graph.add_edge("math", END)
graph.add_edge("general", END)

app = graph.compile()

# 测试不同路由
print(app.invoke({"query": "今天天气怎样？"})["response"])   # → 天气预报...
print(app.invoke({"query": "帮我算 1+1"})["response"])       # → 数学计算...
print(app.invoke({"query": "你好"})["response"])             # → 通用回答...

def route(state):
    # 返回布尔值或枚举，而不是节点名
    return "yes" if state["need_tools"] else "no"

graph.add_conditional_edges("agent", route, {
    "yes": "tools",    # 把 "yes" 映射到 tools 节点
    "no": END,         # 把 "no" 映射到 END
})
# LG8 的 tools_condition 就是这个模式！

class Send:
    """发送到图中特定节点的"包"。

    用于条件边中，动态地用【自定义状态】调用某节点。
    重要：发送的状态可以【不同于】图的主状态——
    这就允许了灵活的 map-reduce 工作流：
    同一节点并行执行多次（不同状态），结果聚合回主状态。

    Attributes:
        node: 目标节点名
        arg: 发送给该节点的状态（可以和主 State 不同！）
    """

# pip install langgraph
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
import operator

class OverallState(TypedDict):
    subjects: list[str]
    jokes: Annotated[list[str], operator.add]   # reducer 聚合所有结果

# 局部状态：每次并行的输入（可以和 OverallState 不同！）
class JokeState(TypedDict):
    subject: str

def continue_to_jokes(state: OverallState):
    """fan-out 函数：为每个 subject 生成一个 Send，并行派发。"""
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]
    # 返回 N 个 Send → generate_joke 节点并发执行 N 次，每次输入不同 subject

def generate_joke(state: JokeState) -> dict:
    """被并发调用的节点。注意输入是 JokeState（局部），不是 OverallState。"""
    # 这里假装调 LLM 生成笑话
    return {"jokes": [f"关于{state['subject']}的笑话"]}

# 构建图
graph = StateGraph(OverallState)
graph.add_node("generate_joke", generate_joke)
graph.add_conditional_edges(START, continue_to_jokes)  # START → fan-out 到多个 Send
graph.add_edge("generate_joke", END)                   # 每个结果 → END，reducer 聚合

app = graph.compile()

result = app.invoke({"subjects": ["猫", "狗", "程序员"], "jokes": []})
print(result["jokes"])
# → ['关于猫的笑话', '关于狗的笑话', '关于程序员的笑话']
# 三个 generate_joke 并发执行，结果被 operator.add reducer 聚合

def call_model(state):
    """agent 节点：调 LLM。"""
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def call_tools(state):
    """tools 节点：执行 LLM 要求的工具。"""
    # 执行 AIMessage 里的 tool_calls...
    return {"messages": [tool_results]}

def should_use_tools(state):
    """条件边路由：判断最后一条 AIMessage 有没有 tool_calls。"""
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:        # 有 → 去执行工具
        return "tools"
    return END                     # 没有 → 模型已给出最终答案，结束

# 图结构：agent 和 tools 形成循环
graph.add_node("agent", call_model)
graph.add_node("tools", call_tools)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_use_tools)  # ← 条件边！
graph.add_edge("tools", "agent")   # 工具执行完回到 agent，形成循环

# 这正是 F3 章的 Agent 循环图！LG8 详解