LangGraph 多 Agent 编排实战

# LangGraph 多 Agent 编排实战 ## 为什么需要 LangGraph LangChain 的链式调用适合线性流程，但真实的 AI 应用往往需要： - 根据中间结果决定下一步（条件路由） - 多个 Agent 协作完成复杂任务 - 循环推理直到满足某个条件 - 人工审核后再继续 LangGraph 用有向图（DAG + 循环）建模这些复杂流程。 ## 安装 ```bash pip install langgraph langchain-openai ``` ## 核心概念 ```python from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated # 1. 定义状态 class AgentState(TypedDict): messages: list next_agent: str iteration: int final_answer: str # 2. 创建图 graph = StateGraph(AgentState) # 3. 添加节点（每个节点是一个函数或 Agent） graph.add_node("researcher", research_node) graph.add_node("writer", writer_node) graph.add_node("reviewer", reviewer_node) # 4. 添加边（包括条件边） graph.add_edge("researcher", "writer") graph.add_conditional_edges("reviewer", review_decision, {"approve": END, "revise": "writer"}) # 5. 设置入口 graph.set_entry_point("researcher") # 6. 编译并运行 app = graph.compile() result = app.invoke({"messages": [user_query], "iteration": 0}) ``` ## 实战：研究-写作-审核系统 ```python from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini") def research_node(state: AgentState) -> AgentState: """研究 Agent: 收集和整理信息""" messages = state["messages"] query = messages[-1] if isinstance(messages[-1], str) \ else messages[-1].content research_prompt = ( "You are a research assistant. Given the topic below, " "provide 5 key facts and insights:\n\n" + query ) response = llm.invoke(research_prompt) messages.append({"role": "assistant", "content": response.content, "agent": "researcher"}) return {"messages": messages, "next_agent": "writer", "iteration": state["iteration"]} def writer_node(state: AgentState) -> AgentState: """写作 Agent: 基于研究结果撰写内容""" research_content = state["messages"][-1]["content"] iteration = state["iteration"] if iteration > 0: feedback = state["messages"][-1].get("feedback", "") write_prompt = ( "Revise the article based on this feedback: " + feedback + "\n\nOriginal research:\n" + research_content ) else: write_prompt = ( "Write a concise article based on this research:\n\n" + research_content ) response = llm.invoke(write_prompt) state["messages"].append({"role": "assistant", "content": response.content, "agent": "writer"}) return {**state, "iteration": iteration + 1} def reviewer_node(state: AgentState) -> AgentState: """审核 Agent: 评估文章质量""" article = state["messages"][-1]["content"] review_prompt = ( "Review this article. Respond with APPROVE if quality is good, " "or REVISE with specific feedback:\n\n" + article ) response = llm.invoke(review_prompt) state["messages"].append({"role": "assistant", "content": response.content, "agent": "reviewer"}) return state def review_decision(state: AgentState) -> str: """根据审核结果决定路由""" last_msg = state["messages"][-1]["content"] if "APPROVE" in last_msg.upper() or state["iteration"] >= 3: return "approve" return "revise" ``` ## 人工介入 (Human-in-the-Loop) LangGraph 支持在关键节点暂停等待人工输入： ```python from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer, interrupt_before=["reviewer"]) # 运行到 reviewer 节点前暂停 config = {"configurable": {"thread_id": "article-1"}} result = app.invoke(initial_state, config) # 人工查看当前状态 current_state = app.get_state(config) print(current_state.values["messages"][-1]["content"]) # 人工确认后继续 app.invoke(None, config) # 从断点继续 ``` ## 并行执行 多个 Agent 同时工作： ```python graph.add_node("search_web", search_web_node) graph.add_node("search_db", search_db_node) graph.add_node("merge", merge_results_node) # 从入口同时走到两个搜索节点 graph.add_edge("start", "search_web") graph.add_edge("start", "search_db") # 两个搜索完成后合并 graph.add_edge("search_web", "merge") graph.add_edge("search_db", "merge") ``` ## 状态持久化 用 SQLite 或 PostgreSQL 保存对话状态： ```python from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string("checkpoints.db") as saver: app = graph.compile(checkpointer=saver) # 状态会自动保存，应用重启后可恢复 ``` ## 调试技巧 ```python # 可视化图结构 from IPython.display import Image Image(app.get_graph().draw_mermaid_png()) # 流式观察每一步 for event in app.stream(initial_state, config): for node_name, output in event.items(): print("Node:", node_name) print("Output:", output) ``` ## 总结 LangGraph 用图结构解决了 AI 应用中复杂工作流的编排问题。对比简单的链式调用，它支持条件路由、循环推理和人工介入，是构建生产级多 Agent 系统的理想框架。