1. 项目概述为什么说 LangChain 和 LangGraph 是 LLM 应用开发中真正能“落地”的两把利剑我从 2023 年初开始在生产环境里跑第一个带记忆的客服对话系统那时候连ChatPromptTemplate都还没进主干分支大家全靠手拼 system/user/assistant 消息、硬写messages.append()、用json.loads()解析 LLM 返回的字符串——结果就是模型一 hallucinate整个流程就卡死在 JSONDecodeError 里日志里全是红色 traceback。直到 LangChain v0.1 推出 Runnable 接口我才第一次意识到原来 LLM 不该被当“黑盒 API”调用而该被当“可组合的函数单元”来编排。两年过去我亲手交付过 7 个不同行业的 LLM 应用保险核保规则解释器、制药企业 SOP 智能检索助手、制造业设备维修知识图谱问答、地方政府政策匹配引擎……所有项目里LangChain 是默认起点LangGraph 是必经拐点。这不是概念炒作而是工程现实——当你需要让大模型“做决定”比如先查数据库、再调外部 API、最后生成报告LangChain 的链式调用会迅速变成面条代码而 LangGraph 的状态机建模能把这种多跳、有分支、需重试的逻辑压缩成一张清晰可测的状态流转图。你不需要成为图论专家但必须理解LangChain 解决的是“怎么把 LLM 和工具连起来”LangGraph 解决的是“怎么让 LLM 和工具按正确顺序、正确条件、正确次数连起来”。热搜词里反复出现的agent、LCEL、langgraph教程本质都是围绕这两个问题展开的实操需求。如果你正在写一个需要调用天气 API 搜索历史订单 生成邮件草稿的智能助理或者要构建一个能自主拆解复杂查询、分步调用多个专业工具的分析系统那这篇内容就是为你写的——它不讲原理推导只讲我在 Ubuntu 22.04 Python 3.11 环境下用miniconda创建隔离环境、部署deepseek-v2本地模型、接入自研维修知识库、最终上线稳定运行 187 天的完整路径。2. 核心设计思路拆解LangChain 与 LangGraph 的分工本质不是“替代”而是“演进”2.1 LangChain 的定位LLM 应用的“基础建设层”解决连接性问题很多人误以为 LangChain 是个“框架”其实它更像一套标准化接口规范。它的核心价值不在功能多而在统一了三类关键抽象Runnable把任何东西LLM、提示词、解析器、甚至普通 Python 函数都包装成.invoke()可调用的对象。这解决了最原始的“胶水代码”问题。比如你写一个get_weather(city: str) - str函数LangChain 要求你把它封装成RunnableLambda(get_weather)这样它就能和ChatOpenAI模型一样被塞进.pipe()链里。我试过直接用原生openai.ChatCompletion.create()结果发现一旦要加重试逻辑、加缓存、加日志埋点就得在每个调用点重复写try/except、redis.get()、logging.info()——而 Runnable 的.with_retry()、.with_config()方法一行代码就全局生效。MessageHistory不是简单存聊天记录而是定义了add_message()、get_messages()的契约。这意味着你可以把 Redis、PostgreSQL、甚至本地 SQLite 当作消息后端只要实现这几个方法上层业务代码完全不用改。我们给某银行做的理财顾问系统初期用InMemoryChatMessageHistory快速验证上线后无缝切换到 Redis只改了两行初始化代码。Tool把外部能力API、数据库查询、文件读取抽象为name、description、args_schema三要素。重点是args_schema——它强制你用 Pydantic 模型声明参数类型和校验规则。这直接堵死了 90% 的前端传参错误。比如一个search_knowledge_base(query: str, top_k: int 3)工具如果前端传top_k: fivePydantic 会在进入工具执行前就抛出ValidationError而不是让 LLM 拿着字符串去数据库里瞎搜。LangChain 的局限也正源于此它默认假设流程是线性的。.pipe()是单向流水线.assign()是单次状态更新。一旦你需要“如果 A 成功则走 B否则走 CB 失败则重试 2 次C 失败则降级为人工”——这套线性范式就崩了。这时候不是 LangChain 不好而是它没设计去处理状态驱动的决策流。2.2 LangGraph 的定位LLM Agent 的“操作系统内核”解决状态与控制流问题LangGraph 的诞生本质上是对 LangChain 线性模型的一次必要补完。它不取代 Runnable而是给 Runnable 加装了“调度器”和“状态寄存器”。它的核心设计哲学只有两条State 是一等公民所有节点node的输入输出都必须是同一个State类型的实例。这个 State 不是全局变量而是每次调用.invoke()时显式传入的快照。比如我们定义class AgentState(TypedDict): messages: list[BaseMessage]; tool_calls: list[dict]; next_action: str那么每个节点函数签名必须是def call_llm(state: AgentState) - AgentState。这强制开发者思考当前步骤改变了哪些字段后续步骤依赖哪些字段避免了隐式状态污染。Graph 是控制流蓝图add_node()注册函数add_edge()定义无条件跳转add_conditional_edges()定义基于 State 字段值的分支判断。最关键的是END节点——它不是退出程序而是告诉调度器“本次循环结束返回当前 State 给调用方”。真正的循环由.stream()或.invoke()的recursion_limit参数控制。我们做过压力测试当recursion_limit25时一个包含 5 个工具调用、3 层嵌套条件判断的 Graph在 16 核 CPU 上平均耗时 1.2 秒失败率低于 0.3%远优于手写 while 循环 if/else 的方案。LangGraph 和 LangChain 的关系就像 Linux 内核和 Shell 命令的关系。ls、grep、curl这些命令对应 LangChain 的 Runnable本身功能强大但要自动化完成“遍历所有日志目录找出含 ERROR 的最新 10 行发邮件给运维组”你就得写 Bash 脚本对应 LangGraph 的 Graph。脚本里依然大量调用ls和grep但控制逻辑循环、条件、错误处理由 Shell 解释器统一管理。LangGraph 正是这个“解释器”。2.3 LCELLangChain 的“函数式编程语法糖”降低组合门槛LCELLangChain Expression Language常被误解为新功能其实它是 LangChain 对 Runnable 组合方式的一次语法升级。它的价值在于把原本需要 5 行代码的链式调用压缩成 1 行可读性强的表达式。比如传统写法from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业客服回答要简洁准确), (user, {input}) ]) llm ChatOpenAI(modelgpt-4-turbo) chain prompt | llm | StrOutputParser()LCEL 写法chain ( ChatPromptTemplate.from_messages([ (system, 你是一个专业客服回答要简洁准确), (user, {input}) ]) | ChatOpenAI(modelgpt-4-turbo) | StrOutputParser() )表面看只是换了个|符号但背后是重大工程改进LCEL 表达式在构建时就完成了类型推断和错误检查。如果你在|后面接了一个返回dict的函数而前面 LLM 返回的是AIMessageLCEL 会在chain.invoke()执行前就报TypeError而不是等到运行时才崩溃。我们团队曾用 LCEL 快速搭建内部文档 QA 系统从写提示词到部署上线只用了 3 小时因为所有类型不匹配都在本地测试阶段暴露了没让一个错误流入测试环境。提示LCEL 不是万能的。当你的链需要动态分支比如根据用户身份切换不同提示词或需要中间状态暂存比如先调用 LLM 生成 SQL再用另一个 Runnable 执行 SQL就必须退回到RunnableLambdaRunnableParallel的组合模式。LCEL 适合线性、确定性流程复杂逻辑请拥抱 LangGraph。3. 实操环境搭建与核心环节实现从零开始部署一个带工具调用的 LangGraph Agent3.1 环境准备用 miniconda 创建纯净、可复现的 Python 环境我坚持用miniconda而非pip全局安装原因很实际LLM 生态的依赖冲突太常见。langchain-core依赖pydantic2.6而langgraph依赖pydantic2.5transformers又要求packaging20.0——手动 pip install 极易陷入“版本地狱”。miniconda的environment.yml文件能锁定所有依赖版本确保团队成员、CI/CD 服务器、生产环境运行完全一致。我的标准环境配置environment.ymlname: langgraph-agent channels: - conda-forge - defaults dependencies: - python3.11 - pip - pip: - langchain0.3.7 - langchain-openai0.2.5 - langgraph0.3.12 - langchain-community0.3.7 - openai1.50.2 - pydantic2.7.4 - redis4.6.0 - psycopg2-binary2.9.9 - jieba0.42.1 # 中文分词必备创建步骤Ubuntu 22.04# 1. 下载并安装 miniconda国内用户建议用清华源 wget https://mirrors.tuna.tsinghua.edu.cn/anaconda/miniconda/Miniconda3-latest-Linux-x86_64.sh bash Miniconda3-latest-Linux-x86_64.sh -b -p $HOME/miniconda3 $HOME/miniconda3/bin/conda init bash source ~/.bashrc # 2. 创建环境并激活 conda env create -f environment.yml conda activate langgraph-agent # 3. 验证关键包版本这是避免后续报错的第一道防线 python -c import langchain; print(langchain.__version__) python -c import langgraph; print(langgraph.__version__)注意不要用conda install langchain直接安装因为 conda-forge 仓库的 langchain 版本通常滞后于 PyPI。务必通过pip在 conda 环境中安装以获取最新修复。3.2 构建核心 Agent一个能查询设备维修知识库并生成处置建议的 Graph我们的目标 Agent 需要完成用户提问 → 判断是否为设备故障类问题 → 若是则搜索维修知识库 → 若找到匹配条目则生成结构化处置建议若未找到则触发人工介入流程。整个过程需支持最多 3 次工具调用且每次调用失败自动重试。第一步定义 State 和 Toolsfrom typing import TypedDict, List, Annotated, Optional from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain_core.pydantic_v1 import BaseModel, Field from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode, tools_condition class AgentState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 自动合并消息列表 tool_calls: List[dict] next_action: str # call_tool, generate_response, escalate retry_count: int # 定义维修知识库查询工具模拟真实数据库查询 class SearchRepairDBInput(BaseModel): query: str Field(description用户问题的关键词如电机过热、PLC通讯中断) top_k: int Field(default3, description返回结果数量) tool(search_repair_db, args_schemaSearchRepairDBInput) def search_repair_db(query: str, top_k: int 3) - List[dict]: 模拟查询设备维修知识库 # 实际项目中这里会连接 PostgreSQL 或 Elasticsearch mock_db [ {id: R001, fault: 电机过热, cause: 散热风扇故障, solution: 清洁风扇叶片检查电机轴承}, {id: R002, fault: PLC通讯中断, cause: 网线松动或交换机故障, solution: 检查物理连接重启交换机}, {id: R003, fault: 传感器读数异常, cause: 传感器校准失效, solution: 执行传感器零点校准流程} ] # 简单关键词匹配真实场景用 BM25 或向量检索 results [item for item in mock_db if query in item[fault] or query in item[cause]] return results[:top_k] # 定义人工介入工具触发工单系统 class EscalateToHumanInput(BaseModel): user_question: str Field(description用户原始问题) context: str Field(description当前已知信息摘要) tool(escalate_to_human, args_schemaEscalateToHumanInput) def escalate_to_human(user_question: str, context: str) - str: 生成人工介入工单 # 实际项目中调用 Jira 或钉钉机器人 API return f已创建工单紧急度-高问题-{user_question}上下文-{context[:50]}...第二步编写 Graph 节点函数from langchain_core.language_models import BaseChatModel from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode # 初始化 LLM此处用 OpenAI实际项目可用 Ollama 本地模型 llm ChatOpenAI(modelgpt-4-turbo, temperature0) # 节点1路由判断决定下一步是调用工具还是生成回复 def route_question(state: AgentState) - str: 根据用户问题判断是否为设备故障类 last_message state[messages][-1] if not isinstance(last_message, HumanMessage): return generate_response # 用 LLM 判断问题类型轻量级分类 prompt ChatPromptTemplate.from_messages([ (system, 你是一个设备维修领域分类器。请判断用户问题是否属于设备故障诊断或维修操作指导。只回答YES或NO。), (user, last_message.content) ]) response prompt | llm | StrOutputParser() result response.invoke({}) return call_tool if result.strip().upper() YES else generate_response # 节点2调用工具使用 LangGraph 内置的 ToolNode tool_node ToolNode([search_repair_db, escalate_to_human]) # 节点3生成最终回复 def generate_response(state: AgentState) - AgentState: 根据工具返回结果生成用户友好的回复 messages state[messages] last_message messages[-1] # 提取工具调用结果从 messages 中找 AIMessage.tool_calls tool_results [] for msg in reversed(messages): if isinstance(msg, AIMessage) and msg.tool_calls: for tc in msg.tool_calls: if tc[name] search_repair_db: tool_results tc[result] elif tc[name] escalate_to_human: return {messages: [AIMessage(contentf已为您创建人工服务工单工单号{tc[result]})]} if not tool_results: return {messages: [AIMessage(content未找到相关维修知识请描述更具体的故障现象。)]} # 用 LLM 整合结果生成回复 prompt ChatPromptTemplate.from_messages([ (system, 你是一名资深设备维修工程师。请根据以下维修知识用中文生成简洁、可操作的处置建议。避免技术术语堆砌。), (user, f故障现象{last_message.content}\n\n参考知识{tool_results}) ]) response prompt | llm | StrOutputParser() final_answer response.invoke({}) return {messages: [AIMessage(contentfinal_answer)]}第三步构建 Graph 并添加条件边# 初始化 Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(route_question, route_question) workflow.add_node(call_tool, tool_node) workflow.add_node(generate_response, generate_response) # 设置入口点 workflow.set_entry_point(route_question) # 添加边route_question 的输出决定流向 workflow.add_conditional_edges( route_question, lambda x: x[next_action], # 这里我们简化实际应返回字符串 { call_tool: call_tool, generate_response: generate_response, } ) # 工具调用后的条件边根据工具执行结果决定下一步 workflow.add_conditional_edges( call_tool, tools_condition, # LangGraph 内置函数自动判断是否还有待执行的 tool_calls { tools: call_tool, # 还有工具要调继续循环 __end__: generate_response, # 工具调用完毕生成回复 } ) # 生成回复后结束 workflow.add_edge(generate_response, END) # 编译 Graph app workflow.compile() # 测试运行 inputs {messages: [HumanMessage(contentPLC通讯中断怎么办)]} for output in app.stream(inputs): for key, value in output.items(): print(f节点 {key}: {value[messages][-1].content})3.3 关键参数与性能调优如何让 Graph 在生产环境稳定运行LangGraph 的recursion_limit和max_concurrency是两个最容易被忽视却影响巨大的参数。recursion_limit默认值是 25但这是指“最大循环次数”不是“最大工具调用次数”。一次循环内可以并行调用多个工具通过RunnableParallel所以实际工具调用总数 recursion_limit×max_concurrency。我们在压测中发现当recursion_limit15时99% 的请求能在 800ms 内完成设为 25 时长尾延迟飙升到 3.2 秒。最终我们定为12并配合超时机制from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph # 使用内存检查点支持中断恢复 checkpointer MemorySaver() app workflow.compile( checkpointercheckpointer, interrupt_before[call_tool], # 在调用工具前可中断 config{ recursion_limit: 12, max_concurrency: 3, } ) # 调用时设置超时 try: result app.invoke( {messages: [HumanMessage(content...)]}, config{configurable: {thread_id: 12345}} ) except Exception as e: # 记录超时日志触发告警 logging.error(fAgent timeout: {e})max_concurrency控制同一循环内并行执行的节点数。设为 1 是严格串行设为os.cpu_count()可能导致线程争抢。我们经过实测在 8 核 CPU 上设为3时吞吐量最高QPS 达 42CPU 利用率稳定在 65% 左右。超过 4 后Redis 连接池开始出现ConnectionResetError说明 I/O 成为瓶颈。实操心得不要迷信“更高并发更好”。我们曾把max_concurrency设为 8结果发现 30% 的请求因 Redis 连接超时失败。后来改用连接池 连接复用并将max_concurrency降为 3错误率归零。记住LLM 应用的瓶颈往往不在 CPU而在网络 I/O 和外部服务响应时间。4. 常见问题与排查技巧实录那些官方文档不会写的“血泪教训”4.1 “State 更新不生效”问题90% 的新手都踩过的坑现象在节点函数里修改了state字典但下一个节点收到的state还是旧值。原因LangGraph 要求节点函数必须返回一个全新的 State 字典而不是就地修改。这是为了保证状态不可变immutability便于调试和回滚。错误写法def bad_node(state: AgentState) - AgentState: state[messages].append(AIMessage(contenthello)) # ❌ 就地修改 return state # 返回的是原引用正确写法def good_node(state: AgentState) - AgentState: new_messages state[messages] [AIMessage(contenthello)] # ✅ 创建新列表 return {messages: new_messages} # ✅ 返回新字典更安全的写法推荐from copy import deepcopy def safe_node(state: AgentState) - AgentState: new_state deepcopy(state) # 深拷贝杜绝副作用 new_state[messages].append(AIMessage(contenthello)) return new_state注意deepcopy有性能开销但在大多数场景下可忽略。如果 State 数据量极大如包含图像 base64 字符串则需手动构造新字典避免深拷贝。4.2 “Tool 调用无限循环”问题条件边配置失误的典型表现现象Graph 进入call_tool节点后反复调用同一个工具直到recursion_limit触发异常。原因tools_condition函数的返回值与add_conditional_edges()中定义的键不匹配。tools_condition默认返回tools表示还有工具要调或__end__表示工具调用完毕。如果你在add_conditional_edges()中写了{tools: call_tool, end: generate_response}那么__end__就找不到对应节点Graph 会默认跳回入口点造成死循环。排查步骤在call_tool节点后加日志打印tools_condition的返回值检查add_conditional_edges()的映射字典确认键名完全一致注意大小写和下划线最稳妥的做法是显式指定tools_condition的返回值def custom_tools_condition(state: AgentState) - str: if state.get(tool_calls): return continue else: return done workflow.add_conditional_edges( call_tool, custom_tools_condition, { continue: call_tool, done: generate_response } )4.3 “中文乱码与 Token 截断”问题模型输入长度控制的硬伤现象用户输入中文问题LLM 返回乱码或“抱歉我无法回答”。原因OpenAI 等模型对输入 token 数有限制gpt-4-turbo 为 128K但 LangChain 的messages会自动拼接如果历史消息过长新问题就会被截断。更隐蔽的是ChatPromptTemplate的format()方法在处理中文时len(prompt.format(...))返回的是字符数而非 token 数导致预估严重偏差。解决方案前置截断在route_question节点前用tiktoken计算总 token 数只保留最近 N 条消息import tiktoken enc tiktoken.encoding_for_model(gpt-4-turbo) def truncate_messages(messages: List[BaseMessage], max_tokens: int 8000) - List[BaseMessage]: total_tokens 0 truncated [] for msg in reversed(messages): # 从最新消息开始保留 tokens len(enc.encode(msg.content)) if total_tokens tokens max_tokens: truncated.append(msg) total_tokens tokens else: break return list(reversed(truncated)) # 恢复时间顺序Prompt 压缩对历史消息做摘要用 LLM 生成一句话总结替换原始长消息。我们用gpt-3.5-turbo做摘要成本极低效果显著。4.4 “Agent 响应慢”问题不是模型慢而是架构没优化现象单次app.invoke()耗时超过 5 秒但单独调用llm.invoke()只需 300ms。排查清单按优先级排序检查 Redis 连接langgraph默认用MemorySaver但生产环境必须用RedisSaver。如果 Redis 服务器延迟高50ms每次状态保存/加载都会拖慢整体速度。用redis-cli --latency测试。禁用冗余日志langgraph的DEBUG日志会记录每一步 State 变化I/O 开销巨大。生产环境务必设为INFO级别。减少消息体积避免在messages中存二进制数据如图片 base64、大段 HTML。用metadata字段存 ID实际数据从数据库查。预热 LLM 连接池首次调用llm.invoke()会有 TLS 握手开销。在应用启动时主动调用一次空请求# 应用启动时 llm.invoke(ping) # 预热连接池最后分享一个小技巧我们给所有 Agent 节点加上traceable装饰器来自langsmith在 LangSmith 平台里能直观看到每个节点的耗时、输入输出、错误堆栈。曾经发现search_repair_db工具里一个正则表达式.*导致回溯爆炸耗时从 120ms 涨到 2.3 秒——这个细节光看代码根本发现不了必须靠可观测性工具。