LangChain核心架构：Chain、Agent、Tool三大组件剖析
引言
去年双十一前夕，我接手了一个客服工单自动处理系统的重构。原系统用硬编码的 if-else 串联了十几个 LLM 调用：先分类工单、再查知识库、然后判断是否需要转人工、最后生成回复。上线三个月后，产品经理要求新增"退款流程自动审批"能力，我打开代码一看——process_ticket() 函数已经膨胀到 800 多行，里面嵌套了 6 层 if-else，每次新增一个业务分支，都要在函数中间小心翼翼地插入代码，稍有不慎就会踩到前面的分支。
这不是个例。几乎所有从"调 API 写 Demo"走向"构建生产系统"的团队，都会撞上同一堵墙：LLM 应用的本质是一次次"决策 + 执行"的循环，而 if-else 是最不适合表达循环和决策的抽象。
LangChain 用三个组件试图解决这个问题：
- Chain：把"固定的多步流程"抽象成可组合的管道
- Tool：把"外部能力"抽象成 LLM 可调用的函数
- Agent：把"决策循环"抽象成由 LLM 驱动的状态机
很多人把这三者当成"三个 API 类"来用，结果写出来的代码比 if-else 还乱。真正理解它们，需要看穿它们背后各自解决的是流程编排、能力边界、决策循环这三件本质不同的事。
本文不打算复述官方文档，而是带你从源码层面看清楚：Chain.invoke() 到底做了什么？AgentExecutor 的 while 循环里藏了哪些坑？Tool 的 schema 为什么能约束住"幻觉"？最后我会给出三个可直接运行的生产级示例，以及一份踩坑清单。
核心概念：从"餐厅"到"车间"
先用一个生活类比把三者的分工讲清楚。
Chain 像餐厅的固定套餐流程：客人点"A 套餐"，后厨就按"前菜 → 主菜 → 甜点"的固定顺序出餐。每一步的输入是上一步的输出，顺序写死在菜单上。这种流程确定性高、延迟低，但客人没法说"今天主菜换成牛排"。
Tool 像后厨的一件专用设备：比如那台只能做意式浓缩的咖啡机。它有一个明确的接口（放咖啡豆、按按钮），有明确的输入输出（豆子进、浓缩出），而且它自己不会主动工作——必须有厨师去操作它。
Agent 像一位能看菜单、会指挥后厨的主厨：客人说"我想吃点清淡的，预算 100"，主厨会自己判断：要不要先看看今天有什么食材？要不要问问客人忌口？然后决定调用哪台设备、按什么顺序做菜。主厨的决策是动态的，是"看一眼情况、做一个动作、再看一眼情况"的循环。
用工程语言重新定义：
| 组件 | 本质 | 控制流归属 | 典型场景 |
|---|---|---|---|
| Chain | 有向无环图（DAG）的静态编排 | 开发者 | 固定的 RAG 流程、文档摘要、格式转换 |
| Tool | 带 schema 的外部函数封装 | 被调用方 | 搜索、数据库查询、API 调用、代码执行 |
| Agent | 由 LLM 驱动的 ReAct 状态机 | LLM（运行时决定） | 开放域问答、多步任务、需要动态规划的场景 |
关键区别在于控制流归属：Chain 的下一步是代码写死的，Agent 的下一步是 LLM 在运行时"想"出来的。这个区别决定了两者在可预测性、延迟、成本上的巨大差异。
下面这张图展示了三者在一个典型 Agent 执行流程中的协作关系：
注意图中从 O 回到 P 的那条线——这就是 Agent 的"心脏"，即 ReAct 循环。理解 Agent，本质上就是理解这个循环的每一次迭代在做什么、什么时候会停下来、什么时候会失控。
源码深度分析
Chain：为什么 invoke() 只是一个"模板方法"
打开 LangChain 的 libs/langchain/langchain/chains/base.py，你会看到 Chain 的核心实现：
class Chain(Serializable, Runnable[Dict[str, Any], Dict[str, Any]], ABC):
@abstractmethod
def _call(
self,
inputs: Dict[str, Any],
run_manager: Optional[CallbackManagerForChainRun] = None,
) -> Dict[str, Any]:
"""执行链的逻辑，子类必须实现。"""
raise NotImplementedError
def invoke(
self, input: Dict[str, Any], config: Optional[RunnableConfig] = None
) -> Dict[str, Any]:
# 1. 参数校验
self._validate_inputs(input)
# 2. 准备回调管理器
callback_manager = CallbackManager.configure(
config.get("callbacks"), self.callbacks
)
run_manager = callback_manager.on_chain_start(
{"name": self.__class__.__name__}, input
)
try:
# 3. 真正的执行逻辑（子类实现）
outputs = (
self._call(inputs, run_manager=run_manager)
if run_manager
else self._call(inputs)
)
except BaseException as e:
run_manager.on_chain_error(e)
raise
else:
run_manager.on_chain_end(outputs)
# 4. 输出校验
final_outputs = self.prep_outputs(inputs, outputs, return_only_outputs=True)
return final_outputs这段代码是一个教科书级的模板方法模式。invoke() 定义了执行骨架（校验 → 回调 → 执行 → 校验 → 返回），把真正的执行逻辑交给子类实现 _call()。
这个设计带来两个重要特性：
第一，可观测性统一。所有 Chain（无论多复杂）的回调、日志、追踪都由基类统一处理。你不需要在每个子类里手动埋点。
第二，Runnable 协议的适配。注意 Chain 继承自 Runnable，这让它可以和 LCEL（LangChain Expression Language）的 | 管道运算符无缝配合。prompt | llm | parser 之所以能工作，是因为每一个组件都实现了 Runnable 接口，而 Chain 只是其中一种实现。
这里有个常被忽视的坑：_validate_inputs 会检查 input_keys，如果输入字典缺了键，会在调用 LLM 之前就抛 ValueError。很多初学者在 LCEL 里混用 Chain 和 Runnable 时，会因为输入输出键不匹配报错，根源就在这里——Chain 对输入输出有显式的键约束，而纯 Runnable 没有。
再往深看一层，SequentialChain 的实现揭示了"管道"的本质：
class SequentialChain(Chain):
def _call(self, inputs, run_manager=None):
known_values = inputs.copy()
for i, chain in enumerate(self.chains):
outputs = chain(known_values, return_only_outputs=True)
known_values.update(outputs)
# 检查是否有变量冲突
if set(outputs) & set(self.input_variables):
raise ValueError(f"变量名冲突: {set(outputs) & set(self.input_variables)}")
return {k: known_values[k] for k in self.output_variables}它就是一个循环：把上一步的输出 merge 进上下文，喂给下一步。没有魔法，就是一个 for 循环加字典合并。这也解释了为什么 SequentialChain 不支持条件分支和并行——它的骨架就是一个线性循环。
Tool：schema 是约束 LLM 的"合同"
Tool 的核心不是"能执行函数"，而是用结构化 schema 把"LLM 的自由文本输出"翻译成"可信的函数调用"。
class BaseTool(RunnableSerializable[Union[str, Dict], Any]):
name: str
description: str
args_schema: Optional[Type[BaseModel]] = None
def invoke(self, input, config=None):
# 如果定义了 Pydantic schema，先做参数校验
if self.args_schema is not None:
if isinstance(input, dict):
input = self.args_schema(**input) # 校验并转换
parsed_input = self._parse_input(input)
return self._run(*parsed_input, ...)
@property
def args(self) -> dict:
"""生成工具描述，喂给 LLM。"""
if self.args_schema is not None:
return self.args_schema.schema()["properties"]
return {}关键在于 args_schema。当你定义一个 Tool 时：
class SearchInput(BaseModel):
query: str = Field(description="搜索关键词")
top_k: int = Field(default=3, description="返回结果数量")
search_tool = StructuredTool.from_function(
func=web_search,
name="web_search",
description="搜索互联网获取最新信息，适用于时效性问题",
args_schema=SearchInput,
)args_schema.schema() 会被序列化成 JSON Schema，注入到给 LLM 的 Prompt 里。LLM 看到的不再是"有个叫 web_search 的函数"，而是：
{
"query": {"type": "string", "description": "搜索关键词"},
"top_k": {"type": "integer", "default": 3, "description": "返回结果数量"}
}这就是"合同"：LLM 必须按这个格式输出 Action Input，Tool 在 invoke 时再用 Pydantic 校验一遍。如果 LLM 输出了 top_k: "three"，Pydantic 会抛 ValidationError，Agent 会把这个错误作为 Observation 反馈给 LLM，让它重新生成。
这个"生成 → 校验 → 反馈"的闭环，是 LangChain 抑制幻觉的核心机制。它不保证 LLM 不出错，但保证错误会被捕获并反馈，而不是静默地传入你的业务函数。
Agent：ReAct 循环的每一次迭代
打开 libs/langchain/langchain/agents/agent.py，AgentExecutor._take_next_step 是整个 Agent 的核心：
def _take_next_step(self, name_to_tool_map, color_mapping, inputs,
intermediate_steps, run_manager):
# 1. 把中间步骤格式化成 LLM 能理解的文本
intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)
# 2. 调用 Agent.plan()，让 LLM 决定下一步
output = self.agent.plan(
intermediate_steps,
callbacks=run_manager.get_child() if run_manager else None,
**inputs,
)
# 3. 如果 LLM 输出的是 AgentFinish，结束循环
if isinstance(output, AgentFinish):
return output
# 4. 否则是 AgentAction，解析出工具名和参数
tool = name_to_tool_map.get(output.tool)
if tool is None:
# 工具不存在，生成错误 Observation
observation = f"工具 {output.tool} 不存在，可用工具: {list(name_to_tool_map.keys())}"
return AgentStep(action=output, observation=observation)
# 5. 执行工具
try:
observation = tool.run(output.tool_input, ...)
except Exception as e:
observation = f"工具执行出错: {e}"
return AgentStep(action=output, observation=observation)而外层的 _call 是一个 while 循环：
def _call(self, inputs, run_manager=None):
intermediate_steps = []
iterations = 0
while iterations < self.max_iterations:
iterations += 1
next_step_output = self._take_next_step(...)
if isinstance(next_step_output, AgentFinish):
return next_step_output.return_values
intermediate_steps.extend(next_step_output)
# 检查是否达到早停条件
if len(next_step_output) == 1:
step = next_step_output[0]
if self.early_stopping_method == "force":
return self.agent.return_stopped_response(...)
return self.agent.return_stopped_response(...)看清楚了吗？Agent 的本质就是一个带最大迭代次数的 while 循环。每次循环：
- 把"历史步骤"拼进 Prompt
- 让 LLM 输出"下一步做什么"
- 如果是
AgentFinish，退出 - 如果是
AgentAction，执行工具，把结果 append 到历史 - 回到第 1 步
这里有几个生产环境的致命细节：
细节一：max_iterations 是硬性护栏。默认值是 15（不同版本有差异）。如果 LLM 陷入"调用搜索 → 结果不满意 → 再搜索"的死循环，15 次之后会强制返回 return_stopped_response。但此时你可能已经烧掉了几十美分的 token。生产环境建议设为 5-8。
细节二：intermediate_steps 会线性增长。每一步的 Action 和 Observation 都拼进 Prompt，第 10 步的 Prompt 长度可能是第 1 步的 10 倍。这是 Agent 成本随步数超线性增长的根源。缓解方案见后文"最佳实践"。
细节三：工具异常不会中断循环。看第 5 步的 except——工具抛异常会被捕获成 Observation 字符串。这看起来是优点（容错），但会带来"错误被 LLM 忽略、反复调用同一个坏工具"的问题。需要在 Observation 里明确标注 [ERROR] 并引导 LLM 换策略。
实战代码
下面三个示例，从简单到复杂，覆盖 Chain、Tool、Agent 的核心用法。
示例一：用 LCEL 构建一个可观测的 RAG Chain
"""
示例一：LCEL 构建 RAG Chain
展示 Runnable 协议、流式输出、以及回调追踪
"""
import os
from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
# ---------- 1. 准备检索器 ----------
# 假设你已有文档列表 docs
docs = [
"LangChain 的 Chain 用于编排固定流程。",
"Agent 由 LLM 驱动决策，适合动态任务。",
"Tool 需要定义 args_schema 来约束 LLM 输出。",
]
vectorstore = FAISS.from_texts(docs, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
# ---------- 2. 定义 Prompt ----------
prompt = ChatPromptTemplate.from_messages([
("system", "你是技术助手。基于以下上下文回答问题，不知道就说不知道。\n\n上下文:\n{context}"),
("human", "{question}"),
])
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ---------- 3. 用 LCEL 组装 Chain ----------
# 注意：LCEL 的 | 运算符本质是 RunnableSequence 的语法糖
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
# 并行执行：一路检索，一路透传问题
{
"context": retriever | RunnableLambda(format_docs),
"question": RunnablePassthrough(),
}
| prompt
| llm
| StrOutputParser()
)
# ---------- 4. 调用并追踪 ----------
if __name__ == "__main__":
# 普通调用
answer = rag_chain.invoke("Agent 和 Chain 有什么区别？")
print(f"[普通调用] {answer}\n")
# 流式调用 —— LCEL 的杀手锏，所有组件自动支持
print("[流式输出] ", end="")
for chunk in rag_chain.stream("Tool 的 schema 有什么用？"):
print(chunk, end="", flush=True)
print()
# 带回调的调用 —— 可观测性
from langchain_core.callbacks import BaseCallbackHandler
class LogHandler(BaseCallbackHandler):
def on_chain_start(self, serialized, inputs, **kwargs):
print(f" >> 进入 Chain: {serialized.get('name', 'unknown')}")
def on_retriever_end(self, documents, **kwargs):
print(f" >> 检索到 {len(documents)} 篇文档")
rag_chain.invoke("什么是 ReAct？", config={"callbacks": [LogHandler()]})关键点：LCEL 的 | 不是普通的管道，它是 RunnableSequence。每个 Runnable 都实现了 invoke/stream/batch 三件套，所以整个 Chain 自动获得流式和批量能力——这是 Chain 基类做不到的。新项目应优先用 LCEL，Chain 子类主要用于兼容旧代码。
示例二：定义带校验的 Tool 并处理幻觉
"""
示例二：定义生产级 Tool
重点：args_schema 约束、错误处理、返回值格式化
"""
from typing import Optional
from pydantic import BaseModel, Field, field_validator
from langchain_core.tools import StructuredTool
import json
# ---------- 1. 用 Pydantic 定义严格的参数 schema ----------
class WeatherInput(BaseModel):
"""查询天气的参数。"""
city: str = Field(description="城市名称，如 '北京'、'Shanghai'")
date: Optional[str] = Field(
default=None,
description="日期，格式 YYYY-MM-DD。不填则查今天",
)
@field_validator("date")
@classmethod
def validate_date(cls, v):
if v is None:
return v
import re
if not re.match(r"^\d{4}-\d{2}-\d{2}$", v):
raise ValueError(f"日期格式错误: {v}，应为 YYYY-MM-DD")
return v
# ---------- 2. 实现工具函数，注意异常处理 ----------
def query_weather(city: str, date: Optional[str] = None) -> str:
"""
真实场景这里会调用天气 API。
关键原则：永远不要在工具里抛未捕获异常，返回结构化错误字符串。
"""
try:
# 模拟 API 调用
if city not in ["北京", "上海", "广州", "深圳"]:
return json.dumps(
{"status": "error", "message": f"暂不支持城市 {city}，支持: 北京/上海/广州/深圳"},
ensure_ascii=False,
)
# 模拟返回
return json.dumps(
{"status": "ok", "city": city, "date": date or "today", "temp": "22°C", "weather": "晴"},
ensure_ascii=False,
)
except Exception as e:
return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False)
# ---------- 3. 封装成 Tool ----------
weather_tool = StructuredTool.from_function(
func=query_weather,
name="query_weather",
description=(
"查询指定城市的天气。适用于用户询问天气、温度、是否下雨等场景。"
"支持城市: 北京、上海、广州、深圳。"
),
args_schema=WeatherInput,
return_direct=False, # 结果是否直接返回用户（不经过 LLM 再加工）
)
if __name__ == "__main__":
# 正常调用
print(weather_tool.invoke({"city": "北京", "date": "2026-09-10"}))
# 触发 Pydantic 校验错误 —— 模拟 LLM 幻觉
try:
weather_tool.invoke({"city": "北京", "date": "明天"})
except Exception as e:
print(f"[校验失败] {type(e).__name__}: {e}")
# 查看喂给 LLM 的 schema
print(f"\n[LLM 看到的 schema]\n{json.dumps(weather_tool.args, ensure_ascii=False, indent=2)}")关键点：
args_schema是强约束。LLM 输出date: "明天"会被 Pydantic 直接拒绝，错误信息反馈给 LLM 让它重试。- 工具内部不抛异常，返回结构化错误。这样 Agent 能把错误当作 Observation 继续推理，而不是整个流程崩溃。
description是给 LLM 看的"使用说明"，要写清楚适用场景和边界。很多 Agent 调错工具，根源是 description 写得太含糊。
示例三：带护栏的生产级 Agent
"""
示例三：生产级 Agent
重点：max_iterations、超时、中间步骤裁剪、成本控制
"""
import time
from typing import List, Tuple
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from example2_weather_tool import weather_tool # 复用示例二的工具
# ---------- 1. 定义一个简单计算工具 ----------
from langchain_core.tools import tool
@tool
def calculator(expression: str) -> str:
"""计算数学表达式，如 '2 + 3 * 4'。只支持四则运算。"""
import re
# 安全校验：只允许数字和运算符
if not re.match(r"^[\d\s\+\-\*/\(\)\.]+$", expression):
return "[ERROR] 表达式包含非法字符，仅支持数字和 + - * / ( )"
try:
result = eval(expression, {"__builtins__": {}}, {})
return str(result)
except Exception as e:
return f"[ERROR] 计算失败: {e}"
tools = [weather_tool, calculator]
# ---------- 2. 构建 Prompt ----------
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content=(
"你是一个严谨的助手。规则：\n"
"1. 只在必要时调用工具，能直接回答的直接回答。\n"
"2. 工具返回 [ERROR] 时，不要重复相同调用，换策略或告知用户。\n"
"3. 最多调用 5 次工具，超出则总结已有信息回答。"
)),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
# ---------- 3. 组装 Agent ----------
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_openai_tools_agent(llm, tools, prompt)
class GuardedAgentExecutor(AgentExecutor):
"""在 AgentExecutor 基础上加超时和成本护栏。"""
max_execution_time: float = 30.0 # 秒
def _call(self, inputs, run_manager=None):
start = time.time()
result = super()._call(inputs, run_manager)
elapsed = time.time() - start
if elapsed > self.max_execution_time:
result["warning"] = f"执行超时 {elapsed:.1f}s"
return result
executor = GuardedAgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5, # 硬性迭代上限
max_execution_time=30, # 超时保护
early_stopping_method="force", # 达到上限时强制返回
handle_parsing_errors=True, # LLM 输出格式错误时自动重试
return_intermediate_steps=True, # 返回中间步骤便于调试
)
if __name__ == "__main__":
# 测试 1：需要多步推理
result = executor.invoke({
"input": "北京今天天气怎么样？如果温度高于 20 度，告诉我 25 * 4 是多少。"
})
print(f"\n[回答] {result['output']}")
print(f"[步数] {len(result['intermediate_steps'])}")
for action, obs in result["intermediate_steps"]:
print(f" - 调用 {action.tool}({action.tool_input}) -> {obs[:80]}")
# 测试 2：触发护栏
result = executor.invoke({
"input": "查一下火星的天气，然后计算 1/0。"
})
print(f"\n[护栏测试] {result['output']}")关键点：
max_iterations=5是成本护栏。默认 15 太宽松，生产环境 5-8 足够覆盖绝大多数任务。handle_parsing_errors=True让 LLM 输出格式错误时自动重试，而不是直接崩溃。return_intermediate_steps=True是调试利器，能看清 Agent 每一步的决策。- 自定义 Executor 加超时：
AgentExecutor本身没有超时机制，长任务可能挂死。子类化是最简单的加护栏方式。
方案对比：LangChain vs 其他流派
| 维度 | LangChain Chain | LangChain Agent | LangGraph | LlamaIndex | 裸调 API |
|---|---|---|---|---|---|
| 控制流 | 静态 DAG | LLM 动态 | 显式状态机 | 静态 + 部分动态 | 完全手写 |
| 可预测性 | 高 | 低 | 中高 | 高 | 取决于实现 |
| 调试难度 | 低 | 高 | 中 | 中 | 低 |
| 循环/分支 | 不支持 | 隐式支持 | 原生支持 | 有限 | 手写 |
| 状态持久化 | 弱 | 弱 | 强（checkpoint） | 中 | 手写 |
| 适合场景 | 固定流程 | 开放任务 | 复杂工作流 | RAG 为主 | 极简需求 |
核心洞察：
- Chain 适合"流程已知"：RAG、文档摘要、格式转换。用 LCEL 写，性能好、可观测、易测试。
- Agent 适合"流程未知"：需要 LLM 现场规划的任务。但要接受它的不确定性，并用护栏兜底。
- LangGraph 是 Agent 的进化形态：当你的 Agent 需要"人机协作""状态持久化""多 Agent 协作"时，直接用 LangGraph，它是显式状态机，比
AgentExecutor的隐式循环可控得多。
- LlamaIndex 在 RAG 上更专精：检索策略、索引结构比 LangChain 丰富。但编排能力不如 LangChain。
- 裸调 API 不是落后：如果你的流程只有 2-3 步且永远不变，直接调 API 反而更简单。框架的价值在于抽象能覆盖复杂度，而不是为了用而用。
一个实用的决策树：
最佳实践与避坑指南
坑 1：把 Agent 当 Chain 用
症状：用 Agent 处理"永远按固定顺序执行"的任务，结果每次调用 LLM 决定下一步，延迟翻倍、成本翻倍、还偶尔走错路。
药方：先问自己"下一步能不能写死"。能写死就用 Chain。Agent 的价值在于处理你无法预判的分支。
坑 2：Tool 的 description 写得太随意
症状：Agent 频繁调错工具，或者在不需要工具时也调用。
药方：description 要写清楚三件事——做什么、什么时候用、什么时候不用。例如：
# 差
description = "搜索"
# 好
description = (
"搜索互联网获取实时信息。"
"适用于：新闻、股价、天气等时效性问题。"
"不适用于：常识问答、数学计算（用 calculator 工具）。"
)坑 3：忽略 intermediate_steps 的 token 膨胀
症状：Agent 跑到第 8 步时，单次 Prompt 已经 8000+ token，成本飙升。
药方：自定义 _prepare_intermediate_steps，只保留最近 N 步的完整内容，更早的步骤只保留 Action 摘要：
def truncate_steps(steps, keep_recent=3):
if len(steps) <= keep_recent:
return steps
old = steps[:-keep_recent]
recent = steps[-keep_recent:]
summary = f"[已完成 {len(old)} 步，摘要: " + \
", ".join(f"{s[0].tool}({s[0].tool_input})" for s in old) + "]"
return [("summary", summary)] + recent坑 4：工具里抛未捕获异常
症状：Agent 运行到一半突然崩溃，用户看到 500 错误。
药方：工具内部永远返回结构化错误字符串，让 LLM 有机会自我修正。参考示例二的写法。
坑 5：没有设置 max_iterations
症状：LLM 陷入死循环，疯狂调用工具，账单爆炸。
药方：max_iterations=5~8，加上 max_execution_time 超时。成本失控是 Agent 最常见的生产事故。
坑 6：Prompt 里没写"何时停止"
症状：Agent 明明已经拿到答案，还继续调用工具"确认"。
药方：系统 Prompt 里明确写"信息足够时直接回答，不要重复调用工具"。同时设置 early_stopping_method="force"。
最佳实践清单
- 优先 LCEL，其次 Chain 子类，最后才用 Agent。能用静态流程解决的，绝不引入动态决策。
- 所有 Tool 必须有
args_schema。没有 schema 的 Tool 等于把幻觉直接放进你的业务函数。 - Agent 必须有护栏三件套：
max_iterations+max_execution_time+handle_parsing_errors。 - 用
return_intermediate_steps=True调试，上线后可改为 False 省 token。 - 复杂工作流用 LangGraph。当你的 Agent 需要状态持久化、人机协作、多 Agent 时，
AgentExecutor的隐式循环会成为负担。 - 给工具加幂等性。Agent 可能重复调用同一个工具，写操作类工具要做好幂等或去重。
- 监控 token 消耗。用 LangSmith 或自建回调，记录每次 Agent 调用的步数和 token，设置告警阈值。
总结
回到开头那个客服系统的故事。重构后，我把系统拆成了三层：
- Chain 层：工单分类、知识库检索、回复生成，用 LCEL 编排，稳定可测。
- Tool 层：退款审批、订单查询、用户画像，每个都有严格的 Pydantic schema。
- Agent 层：只负责"这个工单该走哪条流程"的决策，
max_iterations=5，超时 20 秒。
结果是：延迟从平均 8 秒降到 3 秒，成本降低 60%，新增业务分支从"改 800 行函数"变成"加一个 Tool"。
三个组件，三种抽象，对应三类问题：
- Chain 解决"流程编排"——把已知的步骤串起来，是工程问题。
- Tool 解决"能力边界"——用 schema 定义 LLM 能做什么、不能做什么，是接口问题。
- Agent 解决"动态决策"——让 LLM 在运行时规划，是控制论问题。
理解它们的本质区别，比记住 API 更重要。因为框架会迭代，API 会改名，但"静态流程用 Chain、外部能力用 Tool、动态决策用 Agent"这个判断，五年后依然成立。
延伸思考：当 LLM 的推理能力持续增强，Agent 的"决策循环"会不会逐渐取代 Chain 的"固定编排"？我的判断是——不会完全取代，但边界会移动。就像微服务没有完全取代单体，而是让"该拆的拆、该合的合"成为工程判断。未来的主流形态，可能是"LangGraph 式的混合架构"：主干流程是显式状态机（可控），局部节点用 Agent（灵活）。掌握今天这三个组件，正是理解那个未来的起点。