LangGraph：构建智能工作流的新方式

作者：安娜的信息安全说
1. 什么是 LangGraph？

LangGraph 是一个开源框架，旨在帮助开发者使用语言模型（LLM）构建复杂的、有状态的应用程序。它的核心理念是将应用逻辑表示为 图（Graph），其中节点代表操作或任务，边表示任务之间的依赖关系。这种方式非常适合处理多步骤推理、分支逻辑以及需要记忆状态的场景。
2. 为什么选择 LangGraph？

传统的 LLM 应用通常是线性的：输入 → 模型 → 输出。但在实际业务中，很多任务并不是单一的，而是包含多个步骤、条件分支甚至循环。LangGraph 的优势在于：

图结构灵活性：可以轻松表示复杂的工作流。状态管理：支持在节点之间传递和持久化状态。可扩展性：可以集成外部 API、数据库、工具等。异步执行：提高性能，适合大规模应用。

3. 核心概念

节点（Node）：每个节点是一个操作，可以是调用 LLM、执行代码、访问 API 等。边（Edge）：定义节点之间的依赖关系。状态（State）：在图中流动的数据，可以是用户输入、模型输出或中间结果。执行器（Executor）：负责调度和运行整个图。

4. 应用场景

多步骤问答系统：例如法律咨询，需要先解析问题，再检索法规，最后生成答案。智能客服工作流：根据用户意图动态选择处理路径。数据处理管道：结合 LLM 和传统算法，完成复杂的数据清洗和分析。自动化决策系统：在金融、医疗等领域，结合规则和模型进行决策。

5. 代码示例

1. 安装与基础准备

1. pip install -U langgraph langchain openai tiktoken
2. # 如果用 Anthropic/DeepSeek 等模型，请安装各自 SDK

复制代码

说明：LangGraph 与多数 LLM/工具生态兼容，这里以 LangChain 的 ChatOpenAI 作为示例（你可以替换为任意支持的模型）。

2. 入门示例：三步工作流（分析 → 检索 → 生成）

这个例子演示一个最小化图：把用户问题解析为检索查询，再进行（模拟）检索，最后生成答案。

1. # 语言：Python 3.10+
2. from typing import TypedDict, List
3. from langgraph.graph import StateGraph, END
5. # 1) 定义有状态数据结构（在节点之间传递）
6. class State(TypedDict):
7.     question: str
8.     query: str
9.     docs: List[str]
10.     answer: str
12. # 2) 定义节点函数
13. def analyze_question(state: State) -> State:
14.     q = state["question"]
15.     # 简单规则：提取关键词（真实场景可用 LLM）
16.     state["query"] = q.replace("请", "").replace("怎么", "").strip()
17.     return state
19. def search_docs(state: State) -> State:
20.     # 这里用模拟数据，真实场景你可调用搜索 API 或数据库
21.     query = state["query"]
22.     state["docs"] = [
23.         f"文档1：与『{query}』相关的背景知识。",
24.         f"文档2：『{query}』的常见实践。",
25.     ]
26.     return state
28. def synthesize_answer(state: State) -> State:
29.     docs = state["docs"]
30.     question = state["question"]
31.     combined = "\n".join(docs)
32.     state["answer"] = f"问题：{question}\n——基于检索到的信息——\n{combined}\n\n结论：建议按步骤执行。"
33.     return state
35. # 3) 构建图
36. graph = StateGraph(State)
37. graph.add_node("analyze", analyze_question)
38. graph.add_node("search", search_docs)
39. graph.add_node("answer", synthesize_answer)
41. # 定义边
42. graph.set_entry_point("analyze")
43. graph.add_edge("analyze", "search")
44. graph.add_edge("search", "answer")
45. graph.add_edge("answer", END)
47. # 4) 编译并运行
48. app = graph.compile()
50. # 5) 执行一次
51. if __name__ == "__main__":
52.     result = app.invoke({"question": "请解释怎么用 LangGraph 构建复杂工作流？"})
53.     print("最终答案：\n", result["answer"])

复制代码

3. 接入 LLM 与工具调用（Agent 风格）

下面示例展示：一个节点用 LLM 进行意图识别，另一个节点根据意图调用工具（如计算器），最后再用 LLM 生成总结。

1. from typing import TypedDict, Literal, Dict, Any
2. from langgraph.graph import StateGraph, END
3. from langchain_openai import ChatOpenAI
5. llm = ChatOpenAI(model="gpt-4o-mini")  # 任选模型
7. class AgentState(TypedDict):
8.     user_input: str
9.     intent: Literal["math", "search", "chat"]
10.     tool_result: str
11.     final: str
13. def classify_intent(state: AgentState) -> AgentState:
14.     prompt = f"用户输入：{state['user_input']}\n判断意图：math/search/chat（三选一，直接输出关键词）"
15.     intent = llm.invoke(prompt).content.strip().lower()
16.     if intent not in {"math", "search", "chat"}:
17.         intent = "chat"
18.     state["intent"] = intent  # type: ignore
19.     return state
21. def tool_router(state: AgentState) -> str:
22.     # 动态选择下一个节点
23.     return state["intent"]
25. def math_tool(state: AgentState) -> AgentState:
26.     text = state["user_input"]
27.     # 很简单的表达式解析（真实场景可用安全的解析库）
28.     try:
29.         expr = text.replace("计算", "").strip()
30.         value = eval(expr, {"__builtins__": {}})
31.         state["tool_result"] = f"计算结果：{value}"
32.     except Exception as e:
33.         state["tool_result"] = f"无法计算表达式：{e}"
34.     return state
36. def search_tool(state: AgentState) -> AgentState:
37.     q = state["user_input"]
38.     # 模拟搜索
39.     state["tool_result"] = f"搜索：找到与『{q}』相关的3条结果（模拟）"
40.     return state
42. def chat_fallback(state: AgentState) -> AgentState:
43.     # 无工具，直接返回占位
44.     state["tool_result"] = "普通对话，无需工具。"
45.     return state
47. def summarize(state: AgentState) -> AgentState:
48.     prompt = f"""你是助手。
49. 用户输入：{state['user_input']}
50. 工具结果：{state['tool_result']}
51. 请用简洁中文总结并给出下一步建议。"""
52.     state["final"] = llm.invoke(prompt).content
53.     return state
55. graph = StateGraph(AgentState)
56. graph.add_node("classify", classify_intent)
57. graph.add_node("math", math_tool)
58. graph.add_node("search", search_tool)
59. graph.add_node("chat", chat_fallback)
60. graph.add_node("summarize", summarize)
62. graph.set_entry_point("classify")
63. # 动态路由：从 classify 到 math/search/chat
64. graph.add_conditional_edges("classify", tool_router, {
65.     "math": "math",
66.     "search": "search",
67.     "chat": "chat"
68. })
69. # 三条路径均指向 summarize
70. graph.add_edge("math", "summarize")
71. graph.add_edge("search", "summarize")
72. graph.add_edge("chat", "summarize")
73. graph.add_edge("summarize", END)
75. app = graph.compile()
77. if __name__ == "__main__":
78.     print(app.invoke({"user_input": "计算 21*2+3"}))
79.     print(app.invoke({"user_input": "LangGraph 最新文档在哪？"}))
80.     print(app.invoke({"user_input": "你好，今天心情不错"}))

复制代码

4. 分支与循环（包含重试/校验）

这个示例引入一个验证节点，如果答案不合格则回到生成节点重试（形成一个小环）。

1. from typing import TypedDict
2. from langgraph.graph import StateGraph, END
4. class LoopState(TypedDict):
5.     draft: str
6.     tries: int
7.     ok: bool
9. def draft_answer(state: LoopState) -> LoopState:
10.     state["draft"] = f"草稿 v{state['tries']+1}: 内容待完善。"
11.     state["tries"] += 1
12.     return state
14. def validate(state: LoopState) -> LoopState:
15.     # 简单规则：尝试次数 >= 2 就认为合格
16.     state["ok"] = state["tries"] >= 2
17.     return state
19. def router(state: LoopState) -> str:
20.     return "end" if state["ok"] else "again"
22. graph = StateGraph(LoopState)
23. graph.add_node("draft", draft_answer)
24. graph.add_node("validate", validate)
26. graph.set_entry_point("draft")
27. graph.add_edge("draft", "validate")
28. graph.add_conditional_edges("validate", router, {
29.     "again": "draft",
30.     "end": END
31. })
33. app = graph.compile()
35. if __name__ == "__main__":
36.     result = app.invoke({"draft": "", "tries": 0, "ok": False})
37.     print(result)

复制代码

5. 状态持久化与检查点（内置 Checkpointer）

当你需要 多轮对话 或 长链路恢复，可以使用检查点保存状态。下面用内存检查点（生产建议用 SQLite/Postgres 等持久化）。

1. from typing import TypedDict
2. from langgraph.checkpoint.memory import MemorySaver
3. from langgraph.graph import StateGraph, END
5. class ChatState(TypedDict):
6.     history: list
7.     turn: str
9. def add_turn(state: ChatState) -> ChatState:
10.     state["history"].append(state["turn"])
11.     state["turn"] = ""
12.     return state
14. graph = StateGraph(ChatState)
15. graph.add_node("record", add_turn)
16. graph.set_entry_point("record")
17. graph.add_edge("record", END)
19. checkpointer = MemorySaver()
20. app = graph.compile(checkpointer=checkpointer)
22. if __name__ == "__main__":
23.     # thread_id 标识会话
24.     cfg = {"configurable": {"thread_id": "user_123"}}
26.     print(app.invoke({"history": [], "turn": "你好"}, cfg))
27.     print(app.invoke({"history": [], "turn": "今天的天气"}, cfg))
28.     # 第二次调用会在同一个 thread_id 上追加历史

复制代码

要点：

checkpointer=MemorySaver() 启用检查点。configurable.thread_id 用来标识不同会话。下一次调用无需携带完整历史，LangGraph 会自动从检查点恢复并合并状态。

---

6. 多代理协作（Reviewer + Coder）

示例：两个代理通过有向图协作，一个写代码，一个做评审，不合格则返回修改。

1. from typing import TypedDict
2. from langgraph.graph import StateGraph, END
3. from langchain_openai import ChatOpenAI
5. coder_llm = ChatOpenAI(model="gpt-4o-mini")
6. reviewer_llm = ChatOpenAI(model="gpt-4o-mini")
8. class PRState(TypedDict):
9.     task: str
10.     code: str
11.     review: str
12.     approved: bool
14. def coder(state: PRState) -> PRState:
15.     prompt = f"需求：{state['task']} 请编写简洁、可运行的 Python 示例。"
16.     state["code"] = coder_llm.invoke(prompt).content
17.     return state
19. def reviewer(state: PRState) -> PRState:
20.     prompt = f"请评审以下代码并指出问题（若通过请写：APPROVED）：\n{state['code']}"
21.     review = reviewer_llm.invoke(prompt).content
22.     state["review"] = review
23.     state["approved"] = "APPROVED" in review.upper()
24.     return state
26. def route(state: PRState) -> str:
27.     return "end" if state["approved"] else "fix"
29. def fix_code(state: PRState) -> PRState:
30.     prompt = f"评审意见：{state['review']}\n请据此修复代码：\n{state['code']}"
31.     state["code"] = coder_llm.invoke(prompt).content
32.     return state
34. graph = StateGraph(PRState)
35. graph.add_node("coder", coder)
36. graph.add_node("reviewer", reviewer)
37. graph.add_node("fix", fix_code)
39. graph.set_entry_point("coder")
40. graph.add_edge("coder", "reviewer")
41. graph.add_conditional_edges("reviewer", route, {"end": END, "fix": "fix"})
42. graph.add_edge("fix", "reviewer")
44. app = graph.compile()
46. if __name__ == "__main__":
47.     res = app.invoke({"task": "实现一个斐波那契数列函数", "code": "", "review": "", "approved": False})
48.     print("最终代码：\n", res["code"])
49.     print("评审结果：\n", res["review"])

复制代码

7. 流式输出与异步（边跑边看）

当节点生成较长文本时，你可能希望将中间 token/内容流式推送到前端。LangGraph 支持事件流。下面是一个简单的 streaming 示例。

1. import asyncio
2. from typing import TypedDict
3. from langgraph.graph import StateGraph, END
4. from langchain_openai import ChatOpenAI
6. llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)  # 开启流式
8. class StreamState(TypedDict):
9.     prompt: str
10.     output: str
12. async def gen(state: StreamState) -> StreamState:
13.     chunks = []
14.     async for event in llm.astream(state["prompt"]):
15.         # event 是增量块，具体取决于 SDK
16.         chunks.append(event.content)
17.         # 这里可将 event 推送到前端
18.     state["output"] = "".join(chunks)
19.     return state
21. graph = StateGraph(StreamState)
22. graph.add_node("gen", gen)
23. graph.set_entry_point("gen")
24. graph.add_edge("gen", END)
25. app = graph.compile()
27. async def main():
28.     result = await app.ainvoke({"prompt": "请分点说明 LangGraph 的优势"})
29.     print(result["output"])
31. if __name__ == "__main__":
32.     asyncio.run(main())

复制代码

提示：不同 LLM SDK 的流式接口略有差异，请按你选用的 SDK 文档调整（如 astream_events() / on_token() 回调）。

---

8. 最佳实践建议

状态结构要清晰：使用 TypedDict/Dataclass 明确字段含义，保证节点输入/输出一致。节点纯函数化：尽量让节点只根据输入状态产生输出，方便测试与复用。路由逻辑收敛：复杂条件用单独的路由函数，避免在节点里掺杂太多分支。可观察性：为关键节点加日志/事件推送，便于调试与性能监控。持久化：生产环境使用数据库型 Checkpointer（如 SQLite、Postgres），保证恢复正确。异常与重试：配合循环或重试节点处理不稳定外部依赖（API、工具）。

原文地址：https://blog.csdn.net/nielilijy/article/details/156396469