上一篇把 LangChain 1.0 的建模底座讲完了Chat Models 怎么统一调用、Prompt Template 怎么写、结构化输出怎么收敛。那一篇偏“打字”这篇讲“干活”。标题里这个“核心组件2”我把它定位在从单次调用走向多步协同的那一层——工具、智能体、检索、记忆、状态持久化。说白了就是怎么把一个 Chat Model 变成一套能真正处理复杂任务的系统。这个系列适合两类人。一类是从 LangChain 0.x 往 1.0 迁移的老用户很多 API 看着眼熟但已经悄悄换了行为另一类是刚看完基础篇、想知道工具调用和 Agent 到底怎么串起来的新手。我默认你已经知道 Message、ChatPromptTemplate、with_structured_output 这些东西知道 Runnable 的 invoke/stream/batch 三个基本操作。要是这些还不熟先回去补上一篇。1. LangChain 1.0 组件层次与设计思路1.1 1.0 到底重构了什么很多从 0.x 升上来的人第一反应是“为什么我的 import 全红了”。这不奇怪1.0 把版本政策彻底换了langchain、langchain-core、langgraph、langchain-community 全部统一成 1.x 语义化版本之前那种“核心包还在 0.1周边包已经 0.3”的混乱局面被砍掉了。API 面做了一次大收敛官方明确说要保证 1.x 内部不搞破坏性变更你升级补丁版本不会再提心吊胆。但比版本号更重要的是消息结构标准化。1.0 给所有消息规定了统一的 type 字段和 content block 机制SystemMessage、HumanMessage、AIMessage、ToolMessage 的行为完全对齐。工具调用不再是各家模型各自返回一个 JSON 字符串而是统一成结构化的 tool_calls 字段。这一步的收益要等到你同时接多个模型服务商时才能体会到——换模型只需要换一个类消息格式和工具解析逻辑一行不用动。还有一处很多人没注意AgentExecutor 被明确标记为 legacy。官方文档和所有新示例都在推 LangGraph连 create_agent 这种高层封装也是基于 LangGraph 实现的。这不是单纯的“换了个名字”而是把智能体从“一个固定 ReAct 循环”升级成“可检查、可暂停、可回放的状态机”后面我会专门展开。1.2 核心组件的分层关系我习惯把 1.0 的组件画成五层从下往上分别是层级职责核心类/入口模型层统一接入各家 Chat 模型ChatOpenAI、各种 ChatXxx消息与提示层管理输入输出的消息格式BaseMessage、ChatPromptTemplate工具层把函数变成模型可调用的接口tool、bind_tools、ToolNode检索与记忆层提供外部知识与历史状态Retriever、VectorStore、BaseStore编排层把前面的组件组织成多步流程LangGraph、create_agent、StateGraph这个分层的好处是每层只解决一个问题而且上下层之间通过标准接口通信。比如说工具层不关心你是本地函数还是 HTTP API它只要求你返回一个字符串编排层不关心你的向量库是内存版还是远端集群它只通过 Retriever 接口拿结果。你完全可以只取其中某一层的组件拼进自己的系统没人强制你全家桶。1.3 组件学习路径建议如果只让我推荐三条学习路径我会这么说先做一次单轮工具调用模型 tool bind_tools确认你能看到模型生成 tool_calls再用 create_agent 把它变成多轮智能体加一个 checkpointer 体验状态持久化最后用 StateGraph 手搓一个两节点的图搞清楚智能体本质上是“模型→工具→模型”的循环。检索和记忆其实是给这个循环喂数据的配件等循环跑通之后再接进去排查问题会容易得多。2. 函数调用组件从 tool 到 ToolNode 的完整链路2.1 tool 装饰器与参数模式为什么要用工具因为模型本身算不了数、查不了实时数据。你直接问模型“(1789 43) × 7 等于多少”它可能给你一个煞有介事的错误答案但在计算器工具面前这只是普通字符串处理。工具的本质是把“函数”变成“模型看得懂的接口描述”函数名、参数名、参数类型、每个参数的说明以及这个函数干什么用的 docstring全部会被序列化成 schema 塞进模型请求里。LangChain 1.0 里最常用的写法就是 tool 装饰器。如果函数签名简单直接靠类型注解就能生成参数 schema参数一复杂我建议用 Pydantic 写 args_schema尤其是带默认值、枚举约束、数值范围的情况from pydantic import BaseModel, Field from langchain_core.tools import tool class CalculateInput(BaseModel): expression: str Field(description标准数学表达式例如 (1234)*5) precision: int Field(default2, description保留的小数位数, ge0, le10) tool(args_schemaCalculateInput) def calculate(expression: str, precision: int 2) - str: 执行数学计算返回精确结果。 import math try: result eval(expression, {__builtins__: {}}, math.__dict__) return f{result:.{precision}f} except Exception as exc: return f计算失败: {exc}这里有两个实操经验。第一description 一定要写清楚别偷懒模型能不能正确生成参数很大程度取决于这段文字。第二让工具自己吞掉异常并返回错误信息比直接抛异常好——模型看到返回值是“计算失败: xxx”还能自己纠正表达式再试一次你要是把异常抛给上层这条链路就断了。2.2 bind_tools 到 ToolNode 的调用链路工具定义好了怎么让模型“看见”它答案是把工具绑定到模型上model_with_tools model.bind_tools([calculate, web_search])绑定之后模型收到用户问题不是直接回答而是先输出一个“我想调用 calculate参数是……”的结构化意图这就是 model response 里的 tool_calls 字段。模型只负责“决定调什么”并不负责“执行”。真正执行工具的地方是 ToolNode。它遍历模型返回的 tool_calls逐个调用对应的 tool 函数然后把结果包装成 ToolMessage 塞回消息列表。你只要记得ToolMessage 的 id 必须和 tool_call 的 id 一一对应LangGraph 的 ToolNode 会自动帮你做这个匹配。所以手写循环的人最容易踩的坑就是自己拼 ToolMessage 时 id 对不上模型看到“幽灵工具结果”直接混乱。我自己调试过一个案例模型连续生成三次重复工具调用因为每次接到的 ToolMessage 都是旧的 id它认为结果没到位。换成 ToolNode 之后匹配逻辑由框架保证问题立刻消失。2.3 工具层的错误处理与重试工具调用在真实场景里不是 100% 成功的常见三种失败参数解析失败、函数内部异常、返回内容超长。参数解析失败发生在“模型生成 JSON 不合法”或者“参数不符合 schema 约束”这种建议靠重试机制兜底。LangGraph 的 create_agent 提供了一个很省事的参数 exceptions_to_retry把你想重试的异常类型传进去框架会自动把错误信息回填给模型让它换个方式再生成一次工具调用。函数内部异常就是我刚才说的“吞掉并返回字符串”。返回内容超长则需要你自己控制——工具只返回摘要或截断别把几万字全文塞回上下文。工具层的黄金法则是“模型能看到的一切都必须简洁且结构化”一次工具返回超过 500 个 token 就要警惕了。3. 智能体编排create_agent 与 LangGraph 状态机3.1 为什么 1.0 把智能体交给 LangGraph0.x 时代的 AgentExecutor 本质是一个写死的 ReAct 循环模型想一个动作、执行一个动作、观察结果、再想直到它自己说“结束”。这个循环对付简单任务够用但你没法在中间插入人工确认没法暂停一晚第二天恢复也没法回退到某一步重新跑。生产环境一旦复杂起来需求全是这些“没法”。LangGraph 把同一个流程重构成状态机一个图由节点Node和边Edge组成节点是实际干活的地方边决定干完活下一步去哪。状态是一个不断累积的消息列表每个节点读状态、改状态。你可以用条件边写的逻辑判断“上一步是工具结果就回到模型上一步没有工具调用就结束”。一旦是图就能暂停、恢复、检查中间状态、甚至跳到任意历史节点重新执行——这些能力在 AgentExecutor 里根本不可能。1.0 把 create_agent 直接定位成“生产可用的智能体入口”因为它默认就带了消息累积、工具执行循环、状态持久化这些关键能力不用你自己拼三张图。3.2 用 create_agent 五分钟搭一个带工具能力的智能体假设你已经定义好了 calculate 和 web_search 两个工具代码非常简单from langgraph.prebuilt import create_agent from langgraph.checkpoint.memory import InMemorySaver model load_your_chat_model() # 换成你实际接入的模型实例 tools [calculate, web_search] checkpointer InMemorySaver() agent create_agent( modelmodel, toolstools, system_prompt你是数据分析助手。需要计算时调用 calculate需要查资料时调用 web_search。, checkpointercheckpointer, ) config {configurable: {thread_id: demo-001}} resp agent.invoke( {messages: [{role: user, content: 帮我算一下 3.14159 的平方保留 4 位小数。}]}, configconfig, ) print(resp[messages][-1].content)create_agent 内部帮你做了三件事把用户输入包装成消息状态、把已绑定的工具接成 ToolNode、自动加了一条“模型→工具→模型”的循环边直到模型输出不再带 tool_calls 才停下来。它返回的是一个完整消息列表你只用取最后一条作为最终回复。这里要澄清一个常见误解create_agent 不是“另一个 Agent 框架”它是 LangGraph 的预构建图只是把最常用的单智能体拓扑封装好了。你以后要加人工确认环节、拆成多智能体一样可以从这个图切换到手动图。3.3 检查点Checkpointer与多轮对话状态上面的代码里藏着一个关键参数 checkpointer。没有它每次 invoke 都是独立的模型记不住上一轮说过什么有了它同一条 thread_id 下的所有消息会被保存下来下一次 invoke 自动带上历史 —— 这就是你要的“多轮对话记忆”。我建议你从第一行代码就带上 checkpointer哪怕用 InMemorySaver。原因很简单它不影响任何返回结果但能让你随时用同一个 thread_id 连续问几个问题观察消息是怎么累积的。InMemorySaver 只在进程内存里有效服务重启就没了生产环境换成 SqliteSaver 或 PostgresSaver 这类持久化实现接口一模一样只是存储后端变了。我踩过坑教训是如果只在本地脚本验证InMemorySaver 完全够一旦上 API 服务必须换数据库后端否则用户的对话历史在你每次部署后就“失忆”。3.4 手动构建 StateGraph 的基本模式create_agent 虽然方便但你迟早需要搞明白它背后的图长什么样。手动版一般是这样的from typing import Annotated, Literal, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode class AgentState(TypedDict): messages: Annotated[list, add_messages] def agent_node(state: AgentState): res model.bind_tools(tools).invoke(state[messages]) return {messages: [res]} def should_continue(state: AgentState) - Literal[tools, end]: last state[messages][-1] return tools if getattr(last, tool_calls, None) else end graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, ToolNode(tools)) graph.add_edge(START, agent) graph.add_edge(tools, agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, end: END}) compiled graph.compile()注意 messages 字段的 Annotated[list, add_messages]它告诉 LangGraph 新返回的消息要追加到旧消息后面而不是覆盖。这个细节是状态机的核心每个节点只返回“增量”框架负责维护完整状态。你现在再看 create_agent其实它就是把这个模式封装成了参数化配置。4. 检索组件把知识库塞进对话流程4.1 文本切分是检索质量的命门RAG 的四件套是加载文本、切分成块、向量化、存进向量库。我见过太多人把精力花在“选哪个向量库”上结果检索效果差最后发现病根在切分。切分要回答一个问题一块文本多大、和相邻块重叠多少才能让语义完整且检索精准。LangChain 1.0 里的主力切分器是 RecursiveCharacterTextSplitter它按照段落、句子、字符的优先级递归切分尽量让边界落在语义完整的位置。参数上chunk_size 和 chunk_overlap 不存在“唯一最优解”取决于你的内容形态和模型上下文长度。我做知识库时的经验是通用文档 500 字一块、重叠 80 字起步代码库按函数切不按字数切条款类文档块要短200 字以内避免一块里混进两条不同条款。from langchain_text_splitters import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap80, separators[\n\n, \n, 。, , , ], ) chunks splitter.split_text(long_document)4.2 向量库选型与检索器封装开发阶段我强烈建议直接用 InMemoryVectorStore零部署、零配置代码长这样from langchain_core.vectorstores import InMemoryVectorStore vectorstore InMemoryVectorStore.from_texts( chunks, embeddingembedding_model, ) retriever vectorstore.as_retriever(search_kwargs{k: 4})as_retriever是 1.0 里特别值得记住的接口它把任何向量库都包装成一个统一接口不管底层是内存版还是分布式集群调用方式不变。search_kwargs 里的 k 决定召回几条我一般从 4 开始调k 太小漏信息k 太大则会把噪声灌进上下文。等要上生产再换成外部向量库。这时候特别注意两点一是 embedding 模型不能随便换换了维度对不上、语义空间也变了历史数据要重新灌二是外部库的元数据过滤能力、并发上限、检索延迟都不一样别指望换库是“改一行连接配置”的事。4.3 把检索接到智能体里检索和智能体结合最常见的方式是把检索器封装成一个工具交给智能体让模型自己决定“这个问题要不要查知识库”。这样做的好处是模型可以自主判断缺点是会多一次工具调用的时间开销。tool def retrieve_context(question: str) - str: 从内部知识库检索与问题相关的资料片段。 docs retriever.invoke(question) return \n\n.join(doc.page_content for doc in docs)另一种方式是“预检索注入”把问题先检索一遍把结果直接拼进 system prompt再让模型回答。这种方式更可控适合你明确知道“每次都要查库”的场景。两条路我都跑过经验是面向终端用户的问答用预检索注入更稳因为模型不会因为一句“我有知识库工具”就把该查库的问题直接凭记忆回答了。5. 记忆组件会话状态与多轮上下文管理5.1 短期记忆消息修剪有了 checkpointer消息列表会无限增长很快把上下文窗口塞满。修剪是每一个长对话系统都躲不开的活。LangChain 1.0 提供了 trim_messages它按策略从消息列表里删掉一部分历史同时保留系统提示词from langchain_core.messages import trim_messages trimmer trim_messages( strategylast, token_countermodel, max_tokens2000, include_systemTrue, start_onhuman, ) trimmed trimmer.invoke(messages)注意几个参数的意图strategylast 表示保留最后一批消息token_counter 传入模型实例按真实 token 数量计算include_systemTrue 保证系统提示始终留在开头start_onhuman 防止切到一半从 AI 消息开场。这里容易犯的错是不加 start_on模型回答被拦腰截断格式和语义全乱。5.2 长期记忆BaseStore 与跨会话记忆checkpointer 管的是“同一个 thread_id 内的会话历史”跨会话的用户偏好、事实性信息就得靠更上层的长期记忆。LangChain 1.0 里抽象了 BaseStore一个简单的键值存储接口from langchain_core.store import InMemoryStore user_store InMemoryStore() user_store.put((user, u-001, preferences), keytimezone, valueAsia/Shanghai) v user_store.get((user, u-001, preferences), keytimezone)namespace 可以无限嵌套比如按用户 ID、按会话类型分层。生产环境可以换成带持久化的 Store 实现。我的一般做法是每次对话开始时把该用户长期记忆里的相关内容拉出来作为系统提示的一部分注入对话结束后把本轮发现的新事实写回去。这比把全部历史都塞进上下文便宜得多也靠谱得多。5.3 消息 ID 与手动编辑还有一类更细的需求已经生成的消息被证明是错的要删除或改写。LangChain 1.0 允许用 RemoveMessage 按 ID 删掉指定消息再补一条修正后的消息。这个能力在“模型自我纠错后需要消除错误史”的场景里很有用——如果只是追加一条“刚才我说错了”旧错误消息还留在上下文里模型后续仍然可能被它带偏。不过要提醒一句所有手动编辑都必须在调用模型之前完成改完消息列表再重新 invoke框架会把新状态持久化进 checkpointer。6. 高频故障与排查要点6.1 版本升级的坑从 0.x 升 1.0最常见的报错来自 import 路径变动。0.x 时代的 langchain.llms、langchain.chains 这些路径在 1.0 里基本都消失了模型、向量库、文本切分器被拆到各自的独立包里比如 langchain-openai、langchain-text-splitters。遇到 ImportError先去看官方迁移文档里的路径对照表而不是去百度复制一段老代码。另一个隐蔽问题是传递依赖你把 langgraph 升到 1.0 但 langchain-core 还是 0.3运行时会出各种匪夷所思的兼容报错。我一律建议先跑一遍 pip list | grep langchain把整个生态的版本对齐了再说。6.2 工具参数解析失败的定位方法“模型一直重复调用同一个工具”是智能体场景里出现频率最高的问题。多数情况是模型在流式生成时被没有匹配 id 的 ToolMessage 打断或者工具返回的错误信息不够明确。排查思路是先看模型最后一次生成的 tool_call 长什么样再看对应 ToolMessage 的 id 是否匹配。定位方式很简单打印出消息列表的最后三到五条这个链路一目了然。如果确认是参数 schema 描述不清优先改 Field 里的 description而不是换模型。6.3 流式输出与终端截断用 agent.invoke 时整个智能体跑完才一次性返回用户界面上的体验又慢又卡。LangChain 1.0 的智能体支持 stream流式事件里能区分出 token 增量、工具开始调用、工具返回等不同阶段。我把常用事件整理成了一张速查表现象大概率原因处理方式import 报错老路径或依赖版本不齐用官方迁移对照表改路径对齐所有 langchain 包版本工具调用无限重复tool_call id 不匹配或错误信息不清检查 ToolMessage.id用 ToolNode 自动处理上下文很快超限k 太大或消息未修剪降低召回 k配置 trim_messages 后再注入多轮对话互相记不住没配 checkpointer加 checkpointer线程 id 保持一致流式输出不顺畅串行等待工具批量返回用 stream 事件区分输出类型工具阶段给用户提示6.4 调试三板斧最后分享我自己的调试三板斧。第一板斧是“最小复现”固定一条会触发问题的输入去掉检索、去掉记忆只剩 model tools看它还复现不复现——大部分问题到这里就暴露了。第二板斧是“看原始消息”别只看最终回答把消息列表原样打印出来从用户问题开始一路看到最后一条工具调用的历史全在里面。第三板斧是接 traceLangChain 生态自带的可观测工具能记录每一步的输入输出、token 用量和耗时排查生产问题时比你自己加 print 高效十倍。我调试过最久的一个工具调用 Bug就是在 trace 里看到模型实际生成的一个参数名和 schema 不一致回头改了 description 里的措辞问题立刻消失。说实话1.0 的核心组件拆到最后真正难的从来不是某个 API 怎么调而是怎么把各层组件按你的业务场景组合成一套可控的流程。工具让模型能干活智能体让流程能循环检索让模型有知识记忆让系统有连续性——把这四样装进一个状态机里你手里的就不再是一个聊天接口而是一个有判断、有依据、有记性的处理系统了。
