LangGraph实战：从零构建多步AI工作流与智能体开发指南

最近在尝试把一些重复性的文档处理、数据整理和跨系统查询的工作自动化，一开始用脚本和定时任务勉强能跑，但一旦遇到需要判断、回退、等待外部响应或根据结果动态调整流程的场景，就发现传统的线性脚本根本不够用。要么得写一堆 if-else 硬编码，要么就得拆成多个脚本手工串联，不仅维护麻烦，而且状态跟踪和错误恢复几乎要靠人工盯。

这时候开始关注 AI Agent 框架——不是那种只能单次问答的聊天机器人，而是能按流程执行多步任务、能记住上下文、能根据执行结果自主调整路径的智能体。在对比了几个方案后，LangGraph 吸引我的点在于它用“图”来组织工作流，把任务步骤变成节点，把执行逻辑变成边，这样一来，复杂任务就可以被拆解成可复用、可调试、可扩展的流程单元。

但真正上手时发现，市面上很多教程只讲“怎么跑通 demo”，却很少讲清楚“为什么这样设计”以及“实际项目里怎么避免踩坑”。这篇文章，我就结合自己从零搭建智能体的经历，梳理出一套适合新手的 LangGraph 实战路径，重点不止在功能调用，更在于理解其设计逻辑、掌握可落地的工程化方法。

1. 先搞明白 LangGraph 到底解决了什么痛点

在直接写代码之前，有必要先理解 LangGraph 的设计动机。否则很容易陷入“调通样例但不知道下一步该怎么用”的困境。

1.1 从“单次问答”到“多步工作流”的跨越

大多数人在接触 AI 应用时，第一个遇到的模式是问答式：用户输入一个问题，模型返回一个答案。这种模式对于简单查询很有效，但如果任务需要多个步骤才能完成——比如“帮我查一下上个月的销售数据，对比去年同期，生成总结报告并发邮件给相关负责人”——直接扔给模型很容易出现输出不完整、格式混乱或遗漏步骤的情况。

LangGraph 的核心价值，就是把这类多步任务拆解成一个个可执行的节点，并通过路由逻辑（conditional edges）控制执行流程。每个节点可以是一个 LLM 调用、一个工具调用（比如查数据库、调用 API）、或者一个数据处理函数，节点之间通过状态对象传递信息。

1.2 状态管理：智能体的“记忆”机制

传统脚本在处理多步任务时，往往需要把中间结果保存在变量、文件或数据库中，自己管理状态流转。LangGraph 通过内置的 StateGraph 把状态管理抽象出来，开发者只需要定义状态的结构（比如包含用户查询、当前步骤、已收集的数据、下一步动作等字段），然后在每个节点中读写状态即可。

这样做的好处是：

• 可回溯：整个工作流的执行历史都被保留，方便调试和复盘。
• 可中断与恢复：如果任务执行到一半因为网络或资源问题中断，可以从最后一个成功节点恢复，而不必重头开始。
• 易于扩展：新增一个步骤时，只需要增加一个节点和相应的路由逻辑，不需要重写状态管理代码。

1.3 图结构：把业务流程可视化

用“图”来建模工作流，最直观的好处是可视化。LangGraph 支持导出流程图为图片，这对于复杂业务逻辑的沟通和评审非常有帮助。开发过程中，你可以先画出业务流程图，再直接映射成 LangGraph 的节点和边。

更重要的是，图结构天然支持分支、循环和并行（通过异步节点），比如：

• 分支：根据上一步的结果选择下一步走向（例如，如果查询结果为空，走补救分支；如果有数据，走分析分支）。
• 循环：在满足条件时重复执行某个节点（例如，持续监控直到条件达成）。
• 并行：多个独立任务可以同时执行，最后汇总结果。

这种灵活性是线性脚本难以实现的。

2. 环境准备与最小可行示例

理论讲多了容易抽象，我们直接从一个最简单的例子开始，感受 LangGraph 的基本用法。

2.1 安装与依赖

LangGraph 是 LangChain 生态的一部分，但也可以独立使用。建议新建一个虚拟环境，避免包冲突。

# 创建并激活虚拟环境（可选） python -m venv langgraph-env source langgraph-env/bin/activate # Windows 用 langgraph-env\Scripts\activate # 安装核心包 pip install langgraph langchain-openai

这里安装了langgraph和langchain-openai，后者用于调用 OpenAI 模型。如果你用其他模型（如 Anthropic、本地模型），可以替换成对应的 LangChain 集成包。

2.2 配置模型与基础设置

在代码开头，我们先配置 LLM 和必要的参数。以下示例使用 OpenAI GPT-4，你需要替换成自己的 API Key。

import os from langchain_openai import ChatOpenAI # 设置环境变量（建议使用配置文件或环境变量管理，这里为了演示直接写） os.environ["OPENAI_API_KEY"] = "你的 API Key" # 初始化模型 llm = ChatOpenAI(model="gpt-4")

2.3 构建第一个工作流：对话循环

我们实现一个简单的对话循环：用户输入消息，模型回复，并根据用户是否说“再见”来决定是否结束对话。

首先，定义状态结构。LangGraph 使用 TypedDict 来定义状态 schema：

from typing import TypedDict, List class State(TypedDict): messages: List[str] # 记录对话历史 should_continue: bool # 控制是否继续对话

然后，创建两个节点：一个处理用户输入，一个调用模型生成回复。

from langgraph.graph import StateGraph, END # 定义节点函数 def user_input_node(state: State): user_message = input("你：") state["messages"].append(f"用户：{user_message}") # 如果用户输入“再见”，则停止循环 state["should_continue"] = (user_message != "再见") return state def ai_response_node(state: State): # 将对话历史拼接成上下文 context = "\n".join(state["messages"]) prompt = f"以下是对话历史：\n{context}\n请生成回复：" response = llm.invoke(prompt) state["messages"].append(f"AI：{response.content}") return state

接下来，构建图并设置路由逻辑：

# 创建图构建器 graph_builder = StateGraph(State) # 添加节点 graph_builder.add_node("user_input", user_input_node) graph_builder.add_node("ai_response", ai_response_node) # 设置入口点 graph_builder.set_entry_point("user_input") # 添加边：用户输入后总是执行 AI 回复 graph_builder.add_edge("user_input", "ai_response") # 添加条件边：AI 回复后，根据 should_continue 决定下一步 graph_builder.add_conditional_edges( "ai_response", lambda state: "user_input" if state["should_continue"] else END, ["user_input", END] ) # 编译图 graph = graph_builder.compile()

最后，运行工作流：

# 初始状态 initial_state = {"messages": [], "should_continue": True} # 运行图 for step in graph.stream(initial_state): print(f"当前节点：{list(step.keys())}") if "ai_response" in step: print(f"AI 回复：{step['ai_response']['messages'][-1]}")

这个例子虽然简单，但已经包含了 LangGraph 的核心概念：状态、节点、边和条件路由。你可以通过输入“再见”来结束循环。

3. 进阶实战：构建一个文档查询智能体

单次对话循环还体现不出 LangGraph 的真正威力。我们接下来构建一个更实用的智能体：用户输入一个关于某份文档的问题，智能体先检索相关段落，再生成答案，如果答案不完整，还会主动追问补充信息。

3.1 设计工作流

这个智能体的工作流如下：

1. 接收用户问题。
2. 检索文档：从向量数据库中查找与问题相关的段落。
3. 生成初步答案：基于检索结果生成回答。
4. 判断答案完整性：模型自我评估答案是否足够完整，如果不完整，则生成追问。
5. 根据用户反馈继续或结束：如果追问，则等待用户回复并跳回步骤 2；否则结束。

3.2 实现检索节点

首先，我们需要一个文档库。这里为了简化，使用内存向量数据库（Chroma）并插入示例文档。

from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document # 示例文档 docs = [ Document(page_content="LangGraph 是一个基于图的工作流编排框架，适用于多步 AI 任务。"), Document(page_content="状态管理是 LangGraph 的核心功能，支持复杂流程的暂停与恢复。"), Document(page_content="条件边允许根据上一步的结果动态选择下一步路径。") ] # 初始化向量数据库 embeddings = OpenAIEmbeddings() vectorstore = Chroma.from_documents(docs, embeddings) retriever = vectorstore.as_retriever() def retrieve_node(state: State): question = state["messages"][-1] # 最新用户问题 relevant_docs = retriever.get_relevant_documents(question) state["retrieved_docs"] = [doc.page_content for doc in relevant_docs] return state

3.3 实现答案生成与自评估节点

生成答案后，让模型自我评估是否需要追问。

def generate_answer_node(state: State): context = "\n".join(state["retrieved_docs"]) question = state["messages"][-1] prompt = f"""基于以下文档内容： {context} 问题：{question} 请生成回答，并判断回答是否完整（是否需要更多信息才能完整回答）。如果需要追问，请生成追问问题。""" response = llm.invoke(prompt) # 这里简化处理，实际可以解析模型返回的结构化内容 state["current_answer"] = response.content # 假设模型返回中包含“是否需要追问”的标记 state["need_follow_up"] = "需要追问" in response.content return state def follow_up_node(state: State): if state["need_follow_up"]: # 生成追问 follow_up_question = "请提供更多细节以便我更好地回答。" state["messages"].append(f"AI：{follow_up_question}") return state

3.4 组装完整工作流

现在把各个节点组装起来，并设置条件路由。

graph_builder = StateGraph(State) graph_builder.add_node("retrieve", retrieve_node) graph_builder.add_node("generate_answer", generate_answer_node) graph_builder.add_node("follow_up", follow_up_node) graph_builder.set_entry_point("retrieve") graph_builder.add_edge("retrieve", "generate_answer") graph_builder.add_conditional_edges( "generate_answer", lambda state: "follow_up" if state["need_follow_up"] else END, ["follow_up", END] ) graph_builder.add_edge("follow_up", "retrieve") # 追问后重新检索 graph = graph_builder.compile()

这个智能体已经具备了基本的交互式问答能力。当你问一个模糊的问题时，它会主动追问细节，直到能给出完整答案。

4. 工程化实践：调试、优化与部署

跑通 demo 只是第一步，真正在生产环境使用 LangGraph 还需要考虑很多工程细节。

4.1 调试与日志

LangGraph 提供了详细的执行日志，可以通过设置debug=True开启。

config = {"configurable": {"thread_id": "test_1"}} for step in graph.stream(initial_state, config=config, stream_mode="values"): print(step)

此外，建议在每个节点函数中加入日志记录，方便跟踪状态变化和排查问题。

import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def retrieve_node(state: State): logger.info(f"检索问题：{state['messages'][-1]}") # ... 其余代码

4.2 性能优化

• 异步执行：如果节点中有 IO 密集型操作（如网络请求、数据库查询），可以使用异步节点提高并发性能。

async def async_retrieve_node(state: State): # 异步检索 pass graph_builder.add_node("async_retrieve", async_retrieve_node)

• 缓存：对于重复的检索或计算，可以引入缓存机制，避免重复调用。
• 批量处理：如果需要处理多个独立任务，可以考虑批量调用模型或工具。

4.3 错误处理与重试

在实际应用中，网络波动、模型限流、工具故障等错误很常见。LangGraph 本身不提供自动重试机制，需要开发者自己实现。

可以在节点函数中加入重试逻辑：

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def reliable_llm_call(prompt): return llm.invoke(prompt)

或者，更高级的做法是设计一个错误处理节点，专门捕获异常并根据错误类型决定重试、回退还是终止流程。

4.4 状态持久化

对于长时间运行的任务，状态持久化是必须的。LangGraph 支持将状态保存到数据库或文件中，以便中断后恢复。

# 示例：将状态保存到 JSON 文件 import json def save_state(state, filepath): with open(filepath, 'w') as f: json.dump(state, f) def load_state(filepath): with open(filepath, 'r') as f: return json.load(f)

在生产环境中，你可能需要集成 Redis、PostgreSQL 或专门的流程引擎来管理状态。

4.5 版本管理与测试

当工作流变得复杂后，版本管理和自动化测试变得非常重要。

• 版本管理：使用 Git 管理代码，对工作流定义进行版本控制。
• 测试：为每个节点编写单元测试，并为完整工作流编写集成测试。可以使用 LangGraph 的stream或invoke方法模拟输入，验证输出是否符合预期。

5. 常见问题与避坑指南

根据实践经验，新手在入门 LangGraph 时最容易遇到以下几个问题。

5.1 状态设计过于复杂

状态对象应该只包含工作流需要传递的数据，不要塞入无关字段。一开始尽量保持简洁，随着需求迭代再逐步扩展。

反例：

class OvercomplicatedState(TypedDict): messages: List[str] user_info: Dict session_id: str timestamp: str # ... 很多暂时用不上的字段

正例：

class MinimalState(TypedDict): messages: List[str] next_action: str # 仅包含核心控制字段

5.2 条件边逻辑混乱

条件边的判断函数应该尽量简单，只基于当前状态做布尔判断。复杂的路由逻辑可以拆分成多个节点或使用专门的决策节点。

# 不推荐：在条件边函数中做复杂计算 graph_builder.add_conditional_edges( "some_node", lambda state: "path_a" if complicated_condition(state) else "path_b", ["path_a", "path_b"] ) # 推荐：使用专门的决策节点 def decision_node(state): state["next_node"] = "path_a" if complicated_condition(state) else "path_b" return state graph_builder.add_node("decision", decision_node) graph_builder.add_edge("decision", "{next_node}") # 动态路由

5.3 忽略资源清理

如果工作流中打开了文件连接、数据库连接或网络会话，记得在节点中妥善关闭，或者使用上下文管理器确保资源释放。

5.4 过度依赖 LLM

虽然 LangGraph 常与 LLM 结合使用，但并不是所有节点都需要调用模型。简单的逻辑判断、数据转换、API 调用等，能用普通代码实现的就不要用 LLM，以提高性能和降低成本。

6. 总结：从工具使用到工作流思维

LangGraph 最大的价值，不在于提供了多少现成的 AI 功能，而在于引入了一种“工作流思维”——把复杂的智能任务拆解成可编排、可复用、可观测的步骤。

初学者常犯的错误是试图用一个庞大的提示词让模型完成所有事情，结果往往不可控、难调试。而用 LangGraph 的思路，你可以：

1. 拆解任务：把大任务分解成明确的步骤。
2. 设计流程：用图结构描述步骤之间的逻辑关系。
3. 实现节点：每个节点专注做好一件事。
4. 测试迭代：单独测试每个节点，再集成测试整个流程。

这种模块化的方法，不仅让 AI 应用更可靠，也让团队协作和后期维护变得更容易。当你习惯了这种思维方式后，会发现它不仅适用于 AI 任务，任何复杂的业务流程都可以用类似的思路来设计和实现。

最后提醒一点：LangGraph 是一个强大的框架，但并不是所有场景都需要它。对于简单的单次问答或线性任务，直接调用模型可能更轻量。选择工具时，始终从实际需求出发，避免“为了用框架而用框架”。