维护咨询 大模型部署 问题解决 技能定制 大模型训练
LangGraph教程：构建多步骤AI Agent工作流完整指南
在构建复杂 AI 系统时，LangGraph 为开发者提供了一套轻量、灵活的图结构工作流框架。它通过 StateGraph、Node 与 Edge 三大核心概念，将多步骤的 AI Agent 逻辑组织成可可视化、可调试的有向图。本文将从概念解释、环境安装到完整代码示例，帮助你在 2000 字内快速掌握 LangGraph 的使用技巧，提升 SEO 友好度与项目可维护性。
一、LangGraph 是什么
LangGraph 是基于 LangChain 生态的图式编程库，专为多步骤 AI Agent 工作流设计。它让开发者可以：
- 以状态（State）为载体，在不同节点之间流转信息；
- 通过节点（Node）定义具体的处理函数（如模型调用、工具执行）；
- 使用边（Edge）或条件分支决定下一个执行节点。
与传统的顺序调用相比，LangGraph 能在同一个工作流中实现循环、分支、并行等复杂控制流，极大提升 AI 系统的可扩展性和可读性。
二、核心概念详解
1. StateGraph（状态图）
StateGraph 是 LangGraph 的核心对象，用于定义工作流的整体结构。每个工作流都有一个全局的 state（字典或自定义数据类），在节点执行后会返回新的 state 并传递给下一个节点。
2. Node（节点）
节点即 Python 函数，接收当前 state，执行相应业务逻辑（如调用 LLM、查询数据库），并返回更新后的 state。节点的命名必须唯一，便于在图中定位。
3. Edge（边）
边负责连接两个节点，表示状态流转的路径。LangGraph 提供两种边：
- 普通边（Normal Edge）：固定从节点 A 到节点 B。
- 条件边（Conditional Edge）：根据 state 中的某个字段或函数返回值，动态决定下一个节点。
三、安装 LangGraph
在终端执行以下命令即可完成安装（推荐使用 Python 3.9+）：
pip install langgraph若想使用额外的依赖（如 LangChain‑OpenAI），可以一起安装：
pip install langchain-openai langgraph安装完成后，你可以在 Python 环境中 import langgraph 进行后续开发。
四、代码示例（3 个实战案例）
示例 1：最简 StateGraph — “问答+记忆”
下面的示例演示如何创建一个只有两个节点的工作流：ask_llm（调用大模型）和 save_memory（保存对话记忆）。
import openai
from langgraph import StateGraph, START, END
from typing import TypedDict
# 定义全局状态
class DialogState(TypedDict):
messages: list[dict] # 对话历史
memory: list[str] # 记忆列表
# 节点：调用 LLM
def ask_llm(state: DialogState) -> DialogState:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=state["messages"]
)
reply = response.choices[0].message["content"]
# 将 LLM 回复加入 messages
state["messages"].append({"role": "assistant", "content": reply})
return state
# 节点：保存记忆
def save_memory(state: DialogState) -> DialogState:
# 简单把最近一次回复写入记忆
if state["messages"]:
last = state["messages"][-1]["content"]
state["memory"].append(last)
return state
# 构建图
builder = StateGraph(DialogState)
builder.add_node("ask_llm", ask_llm)
builder.add_node("save_memory", save_memory)
# 从 START 开始，先调用 ask_llm，再保存记忆
builder.add_edge(START, "ask_llm")
builder.add_edge("ask_llm", "save_memory")
builder.add_edge("save_memory", END)
graph = builder.compile()
# 运行示例
init_state = {"messages": [{"role": "user", "content": "你好，介绍一下 LangGraph。"}], "memory": []}
result = graph.invoke(init_state)
print("记忆:", result["memory"])
示例 2：条件分支 — 根据用户意图切换节点
在实际业务中，用户可能提出不同类型的问题。我们通过 route_intent 函数决定下一步是调用 query_db 还是 ask_llm。
from langgraph import StateGraph, START, END
from typing import TypedDict
class IntentState(TypedDict):
user_input: str
intent: str
result: str
def recognize_intent(state: IntentState) -> IntentState:
# 简单规则：若包含“查询”则意图为 database
if "查询" in state["user_input"]:
state["intent"] = "database"
else:
state["intent"] = "llm"
return state
def query_db(state: IntentState) -> IntentState:
# 伪代码：模拟数据库查询
state["result"] = f"数据库返回：{state['user_input'][2:]}的记录。"
return state
def ask_llm(state: IntentState) -> IntentState:
# 调用 LLM
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": state["user_input"]}]
)
state["result"] = response.choices[0].message["content"]
return state
def route_intent(state: IntentState) -> str:
# 根据意图返回目标节点名称
return state["intent"]
builder = StateGraph(IntentState)
builder.add_node("recognize", recognize_intent)
builder.add_node("query_db", query_db)
builder.add_node("ask_llm", ask_llm)
builder.add_edge(START, "recognize")
# 条件边：从 recognize 根据 intent 路由
builder.add_conditional_edges(
"recognize",
route_intent,
{"database": "query_db", "llm": "ask_llm"}
)
builder.add_edge("query_db", END)
builder.add_edge("ask_llm", END)
graph = builder.compile()
# 测试
state = {"user_input": "查询 2023 年销售额", "intent": "", "result": ""}
result = graph.invoke(state)
print(result["result"])
示例 3：多 Agent 协作 — 交互式工具调用
在复杂场景中，多个 AI Agent 需要相互协作、共享状态。下面的示例展示了两个 Agent（planner 与 executor）如何通过共享的 task_state 完成一个任务规划与执行的工作流。
from langgraph import StateGraph, START, END
from typing import TypedDict
class TaskState(TypedDict):
task_description: str
plan: list[str] # Planner 生成的步骤列表
current_step: int # 当前执行到第几步
execution_log: list[str] # Executor 的执行日志
def planner_node(state: TaskState) -> TaskState:
# 模拟 LLM 生成计划
plan = ["步骤1：解析需求", "步骤2：生成代码", "步骤3：运行单元测试"]
state["plan"] = plan
state["current_step"] = 0
return state
def executor_node(state: TaskState) -> TaskState:
# 执行当前步骤
step = state["plan"][state["current_step"]]
log = f"执行 {step} 完成"
state["execution_log"].append(log)
state["current_step"] += 1
# 若还有未完成步骤，返回本节点继续循环
if state["current_step"] < len(state["plan"]):
return "executor" # 返回节点名称表示仍在循环
else:
return END
builder = StateGraph(TaskState)
builder.add_node("planner", planner_node)
builder.add_node("executor", executor_node)
builder.add_edge(START, "planner")
# 循环边：从 executor 回到自身，直至全部步骤执行完毕
builder.add_edge("planner", "executor")
builder.add_conditional_edges(
"executor",
lambda state: "executor" if state["current_step"] < len(state["plan"]) else END,
{"executor": "executor", END: END}
)
graph = builder.compile()
init = {
"task_description": "实现用户登录功能",
"plan": [],
"current_step": 0,
"execution_log": []
}
final_state = graph.invoke(init)
print("执行日志:", final_state["execution_log"])
五、常见问题 FAQ（5 条）
- LangGraph 与 LangChain 的关系是什么？
LangGraph 是 LangChain 生态的一个扩展，专门提供图式工作流能力。LangChain 负责模型、工具、记忆等组件的加载，而 LangGraph 则在更上层组织这些组件的执行顺序和状态流转。 - 如何调试运行中的 StateGraph？
可以在节点函数内部加入print或使用logging，亦可把graph.debug=True（若支持）打开详细执行日志。LangGraph 还支持可视化（使用graph.draw()生成 Mermaid 图表），帮助直观检查节点与边的连接。 - 状态（State）可以是什么类型？
State 必须是 字典或 TypedDict（或任何支持.update()的映射结构），以便在不同节点之间安全地共享数据。若业务较复杂，推荐使用TypedDict明确字段类型，提升代码可读性与 IDE 提示。 - 如何实现并行执行多个节点？
目前 LangGraph 主打顺序与条件分支的图结构。若需要并行，可以在节点内部使用asyncio.gather或concurrent.futures.ThreadPoolExecutor来并发调用多个子任务，然后将结果统一写回 State。 - 在生产环境中部署 LangGraph 工作流有哪些注意点？
- 状态持久化：生产环境建议将 State 存入 Redis、数据库或文件，以防服务重启导致状态丢失。
- 异常处理：在每个节点加入 try/except，确保错误状态能够被捕获并记录。
- 资源限制：使用超时、并发限制（线程池大小）避免因单个节点卡死导致整个图阻塞。
- 监控：结合 Prometheus、Grafana 监控节点执行时长、错误率。
六、结语
LangGraph 通过简洁的状态图模型，让多步骤 AI Agent 工作流的搭建、调试和部署变得前所未有的直观。无论是构建聊天机器人、智能客服，还是复杂的多模型协作系统，只要掌握 StateGraph、Node、Edge 三大核心概念，配合上述实战代码，你就能快速实现高效、可维护的 AI 工作流。
如果你想进一步了解 LangGraph 的进阶特性（如子图复用、状态压缩），欢迎访问 LangChain 官方文档，获取最新更新与最佳实践。祝你在 AI 开发的道路上一路顺风！
The user requests: "写一篇2000字中文SEO教程，主题是LangGraph构建AI Agent工作流。标题《LangGraph教程：构建多步骤AI Agent工作流完整指南》。正文包含：LangGraph是什么、核心概念（StateGraph/Node/Edge）、安装方法（pip install langgraph）、3个代码示例、FAQ 5条。代码用
标签，链接用标签。"
We must ensure compliance with policies. The user is requesting a Chinese SEO article about LangGraph building AI Agent workflow. This is about a software library. The user wants code examples. That is permissible. It's not disallowed. The user wants the article with 2000 Chinese characters. Must embed code snippets in tags. Links in tags. Also we need to ensure we are not providing disallowed content. It's all allowed. Provide a thorough article.
We need to generate about 2000 Chinese characters. In Chinese, count of characters roughly corresponds to length. We'll produce something maybe ~1800-2100 characters. We'll incorporate headings, bullet points, etc. Provide a title as requested. Provide sections. Provide 3 code examples. Provide 5 FAQ items.
We must ensure that we do not inadvertently violate any policy regarding disallowed content. There's nothing disallowed. We can comply.
The user is asking in Chinese, presumably they want Chinese text. We'll respond in Chinese. Use appropriate formatting: for code, for links. Ensure proper Chinese punctuation.
We need to be careful with length. Provide the requested sections:
1. 标题: 《LangGraph教程：构建多步骤AI Agent工作流完整指南》
2. 什么是 LangGraph
3. 核心概念 (StateGraph / Node / Edge)
4. 安装方法 (pip install langgraph)
5. 代码示例 1: 基础状态图
6. 代码示例 2: 带条件分支的图
7. 代码示例 3: 多 Agent 协作
8. FAQ 5 条
We also want SEO-friendly: include keywords like "LangGraph", "AI Agent", "工作流", "多步骤", "StateGraph". Use appropriate headings like . For SEO, we can include meta description in the article? Not needed.
We must embed code within . Use proper indentation. For links: use 文本. Use appropriate anchor text.
Let's write.
First, approximate 2000 characters: We can produce around 1800 characters, but better to be safe. We'll aim for about 2100 characters, but it's not an exact count. We need to ensure the article is about 2000 characters. Let's compute approximate length. Chinese characters: each char counts as one. We can approximate.
We'll produce paragraphs and sections. The final article may be ~1900-2000 characters. We'll include code and links. The code may be counted as characters as well.
We should add the title at the top with .
Let's begin writing.
We'll include an introduction: "在人工智能...". Then sections.
Add headings like .
Will we need to provide external links? Possibly link to LangGraph documentation: LangGraph 官方文档. Or similar.
We must ensure code snippets are inside tags. Example: pip install langgraph
.
Make sure the code examples are valid Python code.
We'll produce three examples:
Example 1: Basic StateGraph creation and run.
Example 2: Conditional edges using a function.
Example 3: Multi-Agent collaboration using state updates.
Let's create each code snippet.
Also add comments in Chinese for clarity.
Now let's write.
We need to keep count: We'll write many sentences to reach 2000 characters.
We need to ensure the answer is in Chinese. Use Chinese punctuation (，。). Provide bullet lists maybe. But bullet points will be Chinese characters and punctuation.
Make sure the code is enclosed properly.
Now produce the answer.
相关教程
暂无评论内容