LangGraph Agent编排框架

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

# 定义状态结构
class AgentState(TypedDict):
    messages: Annotated[list, add]  # 使用 add 操作符实现消息追加
    next_agent: str                  # 下一个要执行的 Agent
    current_step: int                # 当前步骤计数

# 创建状态图
# 参数: 状态类型
graph = StateGraph(AgentState)

from typing import TypedDict, Annotated
from operator import add

# 基本状态定义
class SimpleState(TypedDict):
    query: str
    response: str

# 带 reducer 的状态定义
class RichState(TypedDict):
    # Annotated[list, add] 表示新值会追加到列表，而非替换
    messages: Annotated[list, add]
    # 无 reducer，新值直接覆盖旧值
    current_tool: str
    # 使用自定义 reducer
    scores: Annotated[list, add]

# 自定义 reducer 函数
def merge_dicts(existing: dict, new: dict) -> dict:
    """合并字典，新值覆盖已有键"""
    if existing is None:
        return new
    return {**existing, **new}

class MergedState(TypedDict):
    config: Annotated[dict, merge_dicts]

def analyze_query(state: AgentState) -> dict:
    """分析用户查询的节点"""
    # 读取状态
    messages = state["messages"]
    last_message = messages[-1] if messages else ""

    # 执行计算（如调用 LLM）
    analysis = f"分析结果: {last_message}"

    # 返回状态更新（只返回需要更新的字段）
    return {
        "messages": [f"分析完成: {analysis}"],
        "current_step": state.get("current_step", 0) + 1
    }

def generate_response(state: AgentState) -> dict:
    """生成回复的节点"""
    return {
        "messages": ["这是生成的回复"],
        "next_agent": "end"
    }

# 添加节点到图
# 参数: 节点名称, 节点函数
graph.add_node("analyze", analyze_query)
graph.add_node("generate", generate_response)

# 从 START 到指定节点
graph.add_edge(START, "analyze")

# 从一个节点到另一个节点
graph.add_edge("analyze", "generate")

# 从一个节点到 END（结束）
graph.add_edge("generate", END)

def route_by_intent(state: AgentState) -> str:
    """根据意图路由到不同节点"""
    next_agent = state.get("next_agent", "")
    if next_agent == "search":
        return "search_node"
    elif next_agent == "calculate":
        return "calculate_node"
    else:
        return "generate"

# 添加条件边
# 参数: 源节点, 路由函数, 目标节点映射
graph.add_conditional_edges(
    "analyze",              # 源节点
    route_by_intent,        # 路由函数，返回目标节点名称
    {                       # 可选：路由值到节点名的映射
        "search_node": "search",
        "calculate_node": "calculate",
        "generate": "generate"
    }
)

# 更简洁的条件边（路由函数直接返回节点名）
graph.add_conditional_edges(
    "analyze",
    route_by_intent
    # 不提供映射时，路由函数返回值即为节点名
)

from langgraph.checkpoint.memory import MemorySaver

# 创建内存检查点
memory = MemorySaver()

# 编译图时传入检查点
app = graph.compile(checkpointer=memory)

# 使用 thread_id 区分不同会话
config = {"configurable": {"thread_id": "session-1"}}

# 第一次调用
result1 = app.invoke({"messages": ["你好"], "next_agent": "", "current_step": 0}, config)

# 第二次调用（会延续之前的对话状态）
result2 = app.invoke({"messages": ["请继续"]}, config)

# 获取状态快照
snapshot = app.get_state(config)
print(snapshot.values)  # 当前状态值
print(snapshot.next)    # 下一个要执行的节点

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

# 创建 SQLite 连接
conn = sqlite3.connect("chat_history.db", check_same_thread=False)

# 创建检查点
checkpointer = SqliteSaver(conn)

# 编译
app = graph.compile(checkpointer=checkpointer)

from langgraph.checkpoint.postgres import PostgresSaver

# 使用连接字符串
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:password@localhost:5432/mydb"
)

# 首次使用需要创建表
# checkpointer.setup()

app = graph.compile(checkpointer=checkpointer)

from langgraph.types import interrupt, Command

def human_review_node(state: AgentState) -> dict:
    """需要人工审核的节点"""
    # interrupt() 会暂停图执行，返回值传递给调用者
    human_feedback = interrupt("请审核以下内容是否正确：\n" + str(state["messages"]))

    # 人类反馈通过 Command 传入后，此处获得反馈内容
    return {
        "messages": [f"人工审核结果: {human_feedback}"],
    }

# 添加节点和边
graph.add_node("review", human_review_node)
graph.add_edge("analyze", "review")
graph.add_conditional_edges("review", lambda s: s.get("next_agent", "generate"))

# 编译时启用 interrupt
app = graph.compile(checkpointer=memory, interrupt_before=["review"])

# 执行到 review 节点前会暂停
config = {"configurable": {"thread_id": "review-session"}}
result = app.invoke({"messages": ["需要审核的内容"], "next_agent": "", "current_step": 0}, config)

# 查看当前状态
state = app.get_state(config)
print(state.next)  # 显示下一个要执行的节点

# 提供人工反馈并继续执行
app.invoke(
    Command(resume="审核通过，内容正确"),
    config
)

# 在指定节点执行前暂停
app = graph.compile(checkpointer=memory, interrupt_before=["review"])

# 在指定节点执行后暂停
app = graph.compile(checkpointer=memory, interrupt_after=["review"])

# 也可以同时指定
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["review"],
    interrupt_after=["generate"]
)

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

# 定义子图状态
class SubGraphState(TypedDict):
    query: str
    result: str
    sub_step_count: int

# 创建子图
def process_step_1(state: SubGraphState) -> dict:
    return {"sub_step_count": state.get("sub_step_count", 0) + 1}

def process_step_2(state: SubGraphState) -> dict:
    return {"result": f"处理完成: {state['query']}"}

sub_graph = StateGraph(SubGraphState)
sub_graph.add_node("step1", process_step_1)
sub_graph.add_node("step2", process_step_2)
sub_graph.add_edge(START, "step1")
sub_graph.add_edge("step1", "step2")
sub_graph.add_edge("step2", END)

compiled_sub = sub_graph.compile()

# 定义主图状态
class MainState(TypedDict):
    messages: Annotated[list, add]
    task: str
    final_result: str

def call_subgraph(state: MainState) -> dict:
    """在主图节点中调用子图"""
    result = compiled_sub.invoke({
        "query": state["task"],
        "sub_step_count": 0
    })
    return {"final_result": result["result"]}

# 主图
main_graph = StateGraph(MainState)
main_graph.add_node("subprocess", call_subgraph)
main_graph.add_edge(START, "subprocess")
main_graph.add_edge("subprocess", END)

app = main_graph.compile()

# 基本编译
app = graph.compile()

# 带持久化编译
app = graph.compile(checkpointer=memory)

# 带中断编译
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["review"],
    interrupt_after=[]
)

# 同步调用，返回最终状态
result = app.invoke(
    {"messages": ["你好"], "next_agent": "", "current_step": 0},
    config={"configurable": {"thread_id": "test"}}
)
print(result)  # 最终状态字典

# 流式输出，逐步返回每个节点的执行结果
for event in app.stream(
    {"messages": ["你好"], "next_agent": "", "current_step": 0},
    config={"configurable": {"thread_id": "stream-test"}}
):
    # event 是一个字典 {节点名: 节点输出}
    for node_name, node_output in event.items():
        print(f"节点 {node_name} 输出: {node_output}")

# 流式输出的不同模式
for event in app.stream(
    input_data,
    config=config,
    stream_mode="values"   # 输出完整状态值
):
    print(event)

# stream_mode 选项:
# "values" — 每步输出完整状态
# "updates" — 每步输出状态增量（默认）
# "messages" — 输出 LLM 消息流

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

class SupervisorState(TypedDict):
    messages: Annotated[list, add]
    next_agent: str

def supervisor(state: SupervisorState) -> dict:
    """监督者节点：决定下一步调用哪个 Agent"""
    messages = state["messages"]
    last_msg = messages[-1] if messages else ""

    # 简单路由逻辑（实际中通常由 LLM 决定）
    if "搜索" in last_msg:
        next_agent = "researcher"
    elif "写代码" in last_msg:
        next_agent = "coder"
    elif "计算" in last_msg:
        next_agent = "calculator"
    else:
        next_agent = "FINISH"

    return {"next_agent": next_agent}

def researcher(state: SupervisorState) -> dict:
    """研究 Agent"""
    return {"messages": ["研究结果: 找到了相关信息..."]}

def coder(state: SupervisorState) -> dict:
    """编码 Agent"""
    return {"messages": ["代码已生成: print('hello')"]}

def calculator(state: SupervisorState) -> dict:
    """计算 Agent"""
    return {"messages": ["计算结果: 42"]}

# 路由函数
def route_agent(state: SupervisorState) -> str:
    next_agent = state.get("next_agent", "FINISH")
    if next_agent == "FINISH":
        return END
    return next_agent

# 构建图
graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("coder", coder)
graph.add_node("calculator", calculator)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges("supervisor", route_agent)
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
graph.add_edge("calculator", "supervisor")

app = graph.compile()

# 执行
result = app.invoke({"messages": ["我需要写代码"], "next_agent": ""})

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

class SwarmState(TypedDict):
    messages: Annotated[list, add]
    current_agent: str

def agent_a(state: SwarmState) -> dict:
    """Agent A 可以将控制权交给 B"""
    messages = state["messages"]
    last_msg = messages[-1] if messages else ""

    if "需要 B 处理" in last_msg:
        return {
            "messages": ["Agent A: 移交给 Agent B"],
            "current_agent": "agent_b"
        }
    return {
        "messages": ["Agent A: 我已处理完毕"],
        "current_agent": "FINISH"
    }

def agent_b(state: SwarmState) -> dict:
    """Agent B 可以将控制权交回 A"""
    return {
        "messages": ["Agent B: 处理完成，交回 Agent A"],
        "current_agent": "agent_a"
    }

def swarm_router(state: SwarmState) -> str:
    current = state.get("current_agent", "")
    if current == "FINISH":
        return END
    return current

graph = StateGraph(SwarmState)
graph.add_node("agent_a", agent_a)
graph.add_node("agent_b", agent_b)

graph.add_edge(START, "agent_a")
graph.add_conditional_edges("agent_a", swarm_router)
graph.add_conditional_edges("agent_b", swarm_router)

app = graph.compile()
result = app.invoke({"messages": ["需要 B 处理这个问题"], "current_agent": ""})

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
from operator import add

# 定义状态
class ReActState(TypedDict):
    messages: Annotated[list, add]
    tool_calls: Annotated[list, add]
    final_answer: str

# 模拟 LLM 调用
def call_llm(state: ReActState) -> dict:
    messages = state["messages"]
    last_msg = messages[-1] if messages else ""

    # 简化：根据关键词决定是否调用工具
    if "天气" in last_msg:
        return {"messages": ["需要调用天气工具"], "tool_calls": ["get_weather"]}
    elif "计算" in last_msg:
        return {"messages": ["需要调用计算工具"], "tool_calls": ["calculator"]}
    else:
        return {"messages": ["我可以直接回答这个问题"], "final_answer": "这是最终答案"}

# 模拟工具执行
def execute_tools(state: ReActState) -> dict:
    tool_calls = state.get("tool_calls", [])
    results = []
    for tool in tool_calls:
        if tool == "get_weather":
            results.append("北京今天晴天，25度")
        elif tool == "calculator":
            results.append("计算结果: 42")
    return {
        "messages": [f"工具结果: {r}" for r in results],
        "tool_calls": []  # 清空工具调用列表
    }

# 路由：决定是否需要执行工具
def should_use_tools(state: ReActState) -> str:
    if state.get("tool_calls"):
        return "tools"
    if state.get("final_answer"):
        return END
    return "llm"

# 构建图
graph = StateGraph(ReActState)
graph.add_node("llm", call_llm)
graph.add_node("tools", execute_tools)

graph.add_edge(START, "llm")
graph.add_conditional_edges("llm", should_use_tools, {"tools": "tools", END: END})
graph.add_edge("tools", "llm")  # 工具执行后回到 LLM

app = graph.compile(checkpointer=MemorySaver())

# 执行
result = app.invoke(
    {"messages": ["北京今天天气怎么样？"], "tool_calls": [], "final_answer": ""},
    config={"configurable": {"thread_id": "react-1"}}
)
print(result)

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
from operator import add

class ChatState(TypedDict):
    messages: Annotated[list, add]
    user_intent: str

def classify_intent(state: ChatState) -> dict:
    last_msg = state["messages"][-1] if state["messages"] else ""
    if "帮助" in last_msg or "求助" in last_msg:
        return {"user_intent": "help"}
    elif "订单" in last_msg:
        return {"user_intent": "order"}
    else:
        return {"user_intent": "chat"}

def handle_help(state: ChatState) -> dict:
    return {"messages": ["我是客服助手，请问有什么可以帮助您的？"]}

def handle_order(state: ChatState) -> dict:
    return {"messages": ["请提供您的订单号，我来帮您查询。"]}

def handle_chat(state: ChatState) -> dict:
    return {"messages": ["好的，我明白了。还有其他问题吗？"]}

def route_intent(state: ChatState) -> str:
    intent = state.get("user_intent", "chat")
    return {
        "help": "help_node",
        "order": "order_node",
        "chat": "chat_node"
    }.get(intent, "chat_node")

graph = StateGraph(ChatState)
graph.add_node("classify", classify_intent)
graph.add_node("help_node", handle_help)
graph.add_node("order_node", handle_order)
graph.add_node("chat_node", handle_chat)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_intent)
graph.add_edge("help_node", END)
graph.add_edge("order_node", END)
graph.add_edge("chat_node", END)

app = graph.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "chat-session"}}

# 多轮对话
result1 = app.invoke({"messages": ["我需要帮助"], "user_intent": ""}, config)
result2 = app.invoke({"messages": ["查看我的订单"], "user_intent": ""}, config)

# 1. 为复杂状态使用 TypedDict 并明确 reducer
class RobustState(TypedDict):
    messages: Annotated[list, add]    # 消息追加
    context: dict                     # 覆盖更新
    retry_count: int                  # 覆盖更新

# 2. 条件路由中使用枚举避免拼写错误
from enum import Enum

class AgentName(str, Enum):
    RESEARCHER = "researcher"
    CODER = "coder"
    FINISH = "FINISH"

def router(state) -> str:
    # 使用枚举值
    return AgentName.RESEARCHER

# 3. 设置递归限制防止无限循环
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["review"],
)
result = app.invoke(
    input_data,
    config={
        "configurable": {"thread_id": "safe-session"},
        "recursion_limit": 25  # 最大递归/迭代次数
    }
)

# 4. 使用 stream_mode 调试
for event in app.stream(input_data, config=config, stream_mode="updates"):
    print(event)  # 观察每步的状态变化

# 5. 子图状态转换要显式映射
def call_subgraph(state: MainState) -> dict:
    # 显式映射主图状态到子图状态
    sub_input = {
        "query": state["task"],
        "sub_step_count": 0
    }
    result = compiled_sub.invoke(sub_input)
    # 显式映射子图结果到主图状态
    return {"final_result": result["result"]}

# 6. 人机交互时务必使用 checkpointer
# interrupt 需要 checkpointer 来保存和恢复状态
app = graph.compile(checkpointer=memory, interrupt_before=["approval_node"])

# 获取执行历史
config = {"configurable": {"thread_id": "debug-session"}}
state_history = list(app.get_state_history(config))
for state in state_history:
    print(f"步骤: {state.metadata}, 下一节点: {state.next}")

# 查看图结构（生成 Mermaid 图）
print(app.get_graph().draw_mermaid())