1. 先搞清楚 LangGraph 到底解决什么问题如果你正在接触 AI 大模型应用开发大概率会遇到一个典型问题单次对话模型能回答问题但处理复杂任务时经常显得“记性不好”或“步骤混乱”。比如让模型帮你写一份周报它可能先问你要数据接着就忘了该汇总哪些项目或者多个 AI 子任务之间需要传递状态但用简单链式调用很容易丢上下文。LangGraph 就是专门解决这类问题的框架。它不是另一个大模型而是一个用来编排 AI 工作流的工具库。核心能力是让你用有向图的方式定义多个步骤节点和流转逻辑边尤其适合需要记忆、循环、分支判断的多步任务。和 LangChain 相比LangGraph 更聚焦在“状态管理”和“流程控制”。LangChain 像是一个工具箱提供了各种连接模型、数据库、搜索的工具链而 LangGraph 则是专门设计来处理那些需要反复执行、有条件跳转、多角色协作的 AI 任务。比如客服系统中的多轮对话、代码生成中的迭代调试、数据分析中的分段处理这些场景里LangGraph 能让你更清晰地控制任务流。我一般会建议先明确你的需求如果只是简单调用模型 API用 LangChain 或直接发 HTTP 请求可能更轻量但如果任务需要“记住之前步骤的结果”“根据中间结果决定下一步”“多个 AI 智能体轮流工作”那 LangGraph 的图结构会直观很多。2. 环境准备别在依赖版本上踩坑LangGraph 本身是 Python 库但实际跑通一个智能体需要三块环境Python 环境、大模型接入、以及可选的前端或部署环境。这里我先按最小可运行环境拆解。2.1 Python 环境与核心库LangGraph 强烈建议用 Python 3.8 以上版本。低版本可能会遇到异步语法或类型注解问题。安装时最稳妥的方式是新建虚拟环境# 创建并激活虚拟环境可选但推荐 python -m venv langgraph-env source langgraph-env/bin/activate # Windows 用 langgraph-env\Scripts\activate # 安装核心库 pip install langgraph注意LangGraph 会自动安装 langchain-core 等依赖但如果你需要连接具体模型比如 OpenAI GPT、本地部署的 Ollama 模型、或国内平台模型还要额外装对应的 SDK。例如用 OpenAIpip install openai常见坑点有人直接pip install langchain以为包含了 LangGraph其实这是两个库。LangGraph 虽和 LangChain 生态兼容但需要单独安装。2.2 大模型接入准备LangGraph 本身不提供模型你需要自己准备模型 API 或本地模型。根据你的资源选择云端 API适合快速验证OpenAI GPT、Anthropic Claude、智谱、讯飞等。需要准备 API Key并设置环境变量或直接写在配置里。本地模型适合数据敏感或长期运行用 Ollama、vLLM 等工具部署本地大模型。需要保证机器有足够内存/显存比如 7B 模型至少需要 8GB 可用内存。验证模型是否可调用时先别急着集成到 LangGraph直接用官方 SDK 发一条测试请求。比如 OpenAIfrom openai import OpenAI client OpenAI(api_key你的密钥) # 或通过环境变量 OPENAI_API_KEY 设置 response client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: Hello}] ) print(response.choices[0].message.content)能正常返回再往下走避免把网络问题、密钥问题带到工作流调试中。2.3 资源与权限检查网络环境如果用的海外 API确保网络能稳定访问国内平台模型注意区域限制。磁盘空间本地模型需下载权重文件预留 10GB 以上空间。权限如果你在服务器或容器内运行注意写入日志、输出文件的目录权限。3. 第一个智能体从单步对话到有条件循环LangGraph 的核心是定义图Graph。我们从一个最简单的“问答机器人”开始然后逐步加入状态记忆和循环判断。3.1 定义状态和节点LangGraph 使用状态对象State在节点间传递数据。先定义一个基础状态包含用户输入和模型回复from typing import Dict, Any, List from langgraph.graph import StateGraph, END # 定义状态结构。这里用 TypedDict 更清晰也可以用 Pydantic 模型 from typing import TypedDict class AgentState(TypedDict): user_input: str response: str接着创建图结构并添加第一个节点——调用模型的节点from langgraph.graph import StateGraph, END # 初始化图 builder StateGraph(AgentState) # 定义节点函数接收状态返回更新后的状态 def call_model(state: AgentState) - AgentState: # 这里简化处理实际应调用真实模型 user_text state[user_input] # 模拟模型调用 model_reply f你说了: {user_text} return {response: model_reply} # 将节点添加到图中命名为 model builder.add_node(model, call_model)3.2 设置入口和边现在定义工作流的起点和流向# 设置入口节点从 model 节点开始 builder.set_entry_point(model) # 从 model 节点指向结束节点 builder.add_edge(model, END) # 编译图 graph builder.compile()这时你已经有了一个最简工作流输入用户问题 → 模型回答 → 结束。执行试试# 输入初始状态 result graph.invoke({user_input: 你好吗}) print(result[response]) # 输出: 你说了: 你好吗这个例子虽然简单但验证了图的基本结构节点、状态流转、编译执行。3.3 加入循环判断让对话能多轮进行单次问答不够实用我们加一个判断逻辑如果用户说“继续”就再回答一次否则结束。先修改状态增加对话历史class AgentState(TypedDict): user_input: str response: str history: List[str] # 记录对话历史然后增加一个判断节点决定流程是继续还是结束def should_continue(state: AgentState) - str: # 如果用户输入包含继续则返回 continue 分支否则返回 end 分支 if 继续 in state[user_input]: return continue else: return end重新构建图这次包含条件边builder StateGraph(AgentState) # 添加两个节点模型节点和判断节点 builder.add_node(model, call_model) builder.add_node(check, should_continue) # 注意判断节点也需要包装成 node # 设置入口 builder.set_entry_point(model) # 从 model 节点指向 check 节点 builder.add_edge(model, check) # 条件边根据 check 节点的返回值决定流向 builder.add_conditional_edges( check, should_continue, # 决策函数 { continue: model, # 返回 continue 时跳回 model 节点 end: END } ) graph builder.compile()现在执行多轮测试# 第一轮 state {user_input: 你好, history: []} result graph.invoke(state) print(f模型回复: {result[response]}) print(f当前历史: {result[history]}) # 手动模拟第二轮输入实际中可能来自用户 state2 {user_input: 继续, history: result[history]} result2 graph.invoke(state2) print(f第二轮回复: {result2[response]})这个流程虽然还是手动依次调用但已经体现了 LangGraph 的核心能力状态持久化、条件分支、循环。在实际应用中你可以把用户输入环节也做成节点实现全自动多轮对话。4. 多智能体协作拆解复杂任务单智能体循环适合对话场景但更强大的用途是多智能体协作Multi-Agent。比如一个任务需要分析用户需求 → 写代码 → 检查代码质量这三个步骤由不同特化的 AI 智能体完成。4.1 定义多个智能体节点假设我们有三个智能体角色分析员理解用户需求输出任务规格程序员根据规格写代码审核员检查代码质量提出修改意见先扩展状态结构class MultiAgentState(TypedDict): user_query: str specification: str # 分析员输出 code: str # 程序员输出 review_feedback: str # 审核员输出 current_step: str # 当前执行到哪一步然后定义每个节点的行为def analyst_node(state: MultiAgentState) - MultiAgentState: # 模拟分析员处理 spec f任务规格: 需要实现{state[user_query]}的功能要求代码简洁。 return {specification: spec, current_step: analyst} def programmer_node(state: MultiAgentState) - MultiAgentState: # 模拟程序员根据规格写代码 code f# 实现 {state[specification]}\nprint(Hello World) return {code: code, current_step: programmer} def reviewer_node(state: MultiAgentState) - MultiAgentState: # 模拟审核员检查 feedback 代码基本正确建议添加错误处理。 return {review_feedback: feedback, current_step: reviewer}4.2 构建顺序工作流多智能体协作通常有固定顺序分析 → 编程 → 审核。这种线性流程用简单边连接即可builder StateGraph(MultiAgentState) builder.add_node(analyst, analyst_node) builder.add_node(programmer, programmer_node) builder.add_node(reviewer, reviewer_node) # 设置顺序 builder.set_entry_point(analyst) builder.add_edge(analyst, programmer) builder.add_edge(programmer, reviewer) builder.add_edge(reviewer, END) graph builder.compile()执行这个工作流result graph.invoke({user_query: 文件读取, current_step: start}) print(f最终代码: {result[code]}) print(f审核意见: {result[review_feedback]})4.3 加入条件审核循环如果审核不通过可能需要退回修改。我们加一个判断当审核员给出“需要修改”的反馈时退回给程序员重新编码。修改审核节点让它随机模拟通过或修改实际中根据具体规则判断import random def reviewer_node_with_check(state: MultiAgentState) - MultiAgentState: if random.choice([True, False]): # 50% 概率通过 feedback 审核通过 next_step end else: feedback 需要修改添加注释 next_step retry return {review_feedback: feedback, current_step: reviewer, next_step: next_step}调整图结构支持循环builder StateGraph(MultiAgentState) builder.add_node(analyst, analyst_node) builder.add_node(programmer, programmer_node) builder.add_node(reviewer, reviewer_node_with_check) builder.set_entry_point(analyst) builder.add_edge(analyst, programmer) builder.add_edge(programmer, reviewer) # 条件边根据审核结果决定是结束还是重试 def review_decision(state: MultiAgentState) - str: return state.get(next_step, end) builder.add_conditional_edges( reviewer, review_decision, { end: END, retry: programmer # 退回修改 } )这种设计能让工作流自动处理迭代优化直到满足条件为止。在实际代码生成、内容审核等场景中这种“执行-检查-重试”循环非常实用。5. 生产化考量参数配置、错误处理与部署Demo 能跑通只是第一步真要长期使用还得考虑配置管理、错误处理和部署方式。5.1 模型参数与超时控制实际调用模型时需要设置合理的参数和控制策略。以 OpenAI 为例不要在节点函数里硬编码参数而是通过配置传递from openai import OpenAI import os class ModelConfig: def __init__(self): self.api_key os.getenv(OPENAI_API_KEY) self.model gpt-3.5-turbo self.temperature 0.7 self.max_tokens 1000 self.timeout 30 # 秒 def create_model_node(config: ModelConfig): def model_node(state: AgentState) - AgentState: client OpenAI(api_keyconfig.api_key) try: response client.chat.completions.create( modelconfig.model, messages[{role: user, content: state[user_input]}], temperatureconfig.temperature, max_tokensconfig.max_tokens, timeoutconfig.timeout ) reply response.choices[0].message.content return {response: reply} except Exception as e: # 错误处理记录日志并返回错误信息 return {response: f模型调用失败: {str(e)}} return model_node # 使用时 config ModelConfig() builder.add_node(model, create_model_node(config))关键参数说明temperature控制随机性0-1任务要求确定性结果时设低点如 0.2创意任务设高点如 0.8。max_tokens限制生成长度防止意外消耗。timeout避免网络问题导致长时间卡住。5.2 错误处理与重试机制网络调用、模型服务都可能临时失败。对于重要任务需要加入重试逻辑import time from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def reliable_model_call(client, messages, config): return client.chat.completions.create( modelconfig.model, messagesmessages, temperatureconfig.temperature, max_tokensconfig.max_tokens, timeoutconfig.timeout )然后在节点函数中调用这个带重试的版本。注意重试适合临时性错误如网络超时如果是认证失败、配额不足等永久性错误重试只会浪费资源。5.3 状态序列化与持久化长时间运行的工作流可能需要中断后恢复。LangGraph 的状态可以通过json.dumps()序列化保存到数据库或文件import json # 执行工作流 state {user_input: 长时间任务} result graph.invoke(state) # 保存状态例如到数据库 serialized_state json.dumps(result) # db.save(task_123, serialized_state) # 恢复执行 # saved_state db.load(task_123) # loaded_state json.loads(saved_state) # new_result graph.invoke(loaded_state)这对于需要运行几分钟甚至几小时的复杂任务特别重要。5.4 部署方式选择根据使用场景选择部署方案本地脚本适合个人使用或测试直接python your_agent.py运行。Web API用 FastAPI 或 Flask 包装成 HTTP 服务供其他系统调用。异步任务队列如果处理耗时较长用 Celery Redis 等队列系统避免阻塞 Web 请求。云函数无服务器部署按需执行适合低频但需要高可用性的场景。最简单的 FastAPI 示例from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Request(BaseModel): user_input: str app.post(/chat) async def chat_endpoint(request: Request): result graph.invoke({user_input: request.user_input}) return {response: result[response]}6. 性能优化与排查指南智能体工作流跑起来后下一步就是优化性能和稳定性。我从实际项目经验里总结了几条必查项。6.1 减少不必要的 Token 消耗大模型 API 按 Token 收费工作流中多次调用模型时要避免重复发送相同内容。优化策略历史摘要多轮对话中不要每次都发送完整历史而是定期生成摘要。上下文压缩只传递当前步骤需要的上下文而不是整个状态。条件执行某些节点可能不需要每次都被执行通过条件判断跳过。示例只在历史超过 5 轮时才生成摘要def should_summarize(state: AgentState) - bool: return len(state[history]) 5 def summarize_history(state: AgentState) - AgentState: if should_summarize(state): # 调用模型生成摘要 summary 对话摘要... # 实际调用模型生成 return {history: [summary]} # 用摘要替换详细历史 return state # 不需要摘要时原样返回6.2 并发处理与批量优化如果有大量独立任务需要处理可以考虑并发执行。但要注意模型 API 的速率限制import asyncio from langgraph.graph import StateGraph async def process_batch(tasks: List[AgentState]) - List[AgentState]: # 注意控制并发数避免触发 API 限制 semaphore asyncio.Semaphore(5) # 最大并发 5 async def process_one(state: AgentState): async with semaphore: return await graph.ainvoke(state) return await asyncio.gather(*[process_one(task) for task in tasks])批量处理时还要考虑错误隔离一个任务失败不应该影响其他任务。6.3 监控与日志排查生产环境必须要有完善的日志记录。在每个节点开始和结束时记录状态import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def logged_node(node_name): def decorator(node_func): def wrapper(state: AgentState) - AgentState: logger.info(f节点 {node_name} 开始执行输入: {state}) try: result node_func(state) logger.info(f节点 {node_name} 完成输出: {result}) return result except Exception as e: logger.error(f节点 {node_name} 执行失败: {str(e)}) raise return wrapper return decorator # 使用装饰器 logged_node(model) def call_model(state: AgentState) - AgentState: # ... 节点逻辑关键监控指标每个节点的执行时间模型调用的 Token 使用量工作流整体成功率错误类型分布6.4 常见问题排查顺序当工作流出现问题时按这个顺序排查检查输入数据格式是否正确内容是否完整编码有无问题验证模型连接直接调用模型 API 是否能正常返回检查节点顺序通过日志看每个节点的输入输出找到第一个出现异常的节点。审查状态流转条件边的判断逻辑是否按预期工作资源限制检查内存、网络、API 配额是否充足版本兼容性LangGraph、LangChain、模型 SDK 版本是否兼容特别是升级库版本后要重新测试核心工作流避免接口变更导致的问题。7. 什么时候该用 LangGraph什么时候考虑替代方案LangGraph 不是万能解决方案根据你的具体需求选择合适的工具。7.1 适合 LangGraph 的场景复杂多步任务需要多个 AI 调用且步骤间有状态依赖。循环审批流程如内容生成→审核→修改→再审核的循环。多角色协作不同的 AI 智能体负责不同专业领域。需要持久化状态长时间运行的任务需要保存中间结果。条件分支丰富根据中间结果决定下一步走向。7.2 可能过度设计的场景简单问答机器人单轮对话用 LangChain 的 Chain 或直接调用模型更简单。一次性脚本不需要状态管理和复杂流程的简单任务。性能极致要求LangGraph 的抽象层会带来一些开销对延迟极其敏感的场景可能要考虑更底层的实现。7.3 替代方案对比LangChain Expression Language (LCEL)适合线性链式调用语法更简洁但复杂流程表达能力有限。直接编码对于固定流程直接用 Python 函数调用可能更直接可控。专业工作流引擎如 Apache Airflow、Prefect适合调度型任务但 AI 集成需要更多自定义。选择原则从简单方案开始只有当代码中出现了大量的状态传递和条件判断时才考虑引入 LangGraph。我个人经验是先用简单方式实现核心功能当发现状态管理变得混乱时就是引入 LangGraph 的好时机。不要一开始就追求最复杂的架构特别是对于探索性项目。最后提醒一点LangGraph 还在快速发展中API 可能会有调整。生产项目中使用时要锁定依赖版本并有计划地测试新版本。关注官方文档和 GitHub 仓库的更新及时了解新特性和最佳实践变化。