大概每个把LangChain用了一段时间的人都会经历这个阶段：LLMChain、ConversationChain 这些内置链跑起来很顺手，可一旦任务变成“先清洗一批原始文本，再做结构化抽取，最后根据中间结果决定调哪些工具”，默认组件就撑不住了。最开始我试图把所有逻辑塞进一个大 prompt，指望模型一口气输出完整结果，结果不是字段缺失就是答非所问，token 还烧得飞快。后来我转向自定义 Chain 和 Agent 的组合：把确定性的流程拆成固定管线，把不确定的取舍交给 Agent 动态调度，复杂任务才真正有了解法。这篇文章我不讲入门概念，直接记录怎么拆解任务、怎么写自定义 Chain、怎么开发多工具 Agent，以及中间踩过的坑，适合已经跑通 LangChain 基本流程、正要接真实业务场景的开发者参考。

1. 整体设计思路：复杂任务为什么要“自定义”

1.1 复杂任务的两个共性特征

我复盘过自己接过的几个实战需求，发现所谓“复杂任务”基本离不开两件事：固定多阶段操作，加上需要临场判断的决策点。

举个例子，做一个文档分析工具：输入一篇非结构化的业务报告，系统要先清理噪声段落、抽取关键字段、生成摘要，再根据摘要内容决定是否深挖某个主题。前半段“清洗—抽取—摘要”是确定的处理流水线，每一步用什么 prompt、输出什么格式都是可预期的，适合写成 Chain；后半段“是否深挖、调哪个工具、需要什么额外信息”没有固定答案，模型要根据实际情况选择路径，适合交给 Agent。

这种“固定流程 + 动态抉择”的组合，恰恰是内置链很难覆盖的。LLMChain 只能做单次调用的包装，ConversationChain 只管理多轮记忆，它们没有一个机制去编排“先执行 A、再执行 B、然后根据 B 的结果选择 C 或 D”。强行用一个大 prompt 模拟这个过程，模型会陷入两个问题：一是输出结构不稳定，长任务下很容易漏掉某个中间字段；二是中间结果不可校验，你无法在第二步之前检查第一步的质量。

所以自定义的价值并不在于“代码写得好看”，而在于把任务的可控部分和智能部分拆开：可控部分交给 Chain 保证稳定，智能部分交给 Agent 保留灵活性。

1.2 什么时候该用 Chain，什么时候该上 Agent

我自己常用的判断标准就一条：如果“下一步做什么”在任务设计时就能穷举，就用 Chain；如果“下一步做什么”要由模型根据中间结果现想，就用 Agent。别急着上 Agent，Agent 看起来更聪明，但每次多一轮工具调用就多一层出错概率和延迟。

更细致的判断可以参考下面这个表：

我见过不少团队一上来就对所有需求上 Agent，结果一个很简单的“把输入规范化再输出”也走思考—行动—观察循环，耗时翻倍还经常因为格式解析失败重试。反过来，有些任务明明需要基于搜索结果不断追问，却硬写成一长串 Chain，中途一旦某个环节输出偏了，后面全偏。理解这个边界比学会写代码更重要，因为它决定了系统复杂度整体可控。

1.3 一条常用的设计原则：管线负责“会做”，Agent 负责“会想”

我最终沉淀下来的套路是这样的：先把整个任务画成一张流程图，不需要画得很细，就标出哪些框是固定动作、哪些框是决策点。固定动作做成 Chain 节点，决策点做成 Agent 的可选工具，然后把有限的几个决策点连接起来。

这里有个关键纪律：凡是 LLM 不可靠的事情，不要用它做。比如精确计算、格式转换、时间标准化，这些都应该抽成普通 Python 函数放进 Chain，而不是让模型去做。模型擅长的是理解、归纳、生成、判断，把这些能力放在 Agent 的决策层，把确定性的数据处理放在 Chain 层，整个系统才会又快又稳。

另外，模块化也要考虑可测试性。Chain 的每个节点最好都能单独跑通，给定样例输入，能断言输出是否符合预期；Agent 的工具则应该是纯函数式的，输入输出都有明确类型，不要在工具内部偷偷依赖外部全局状态。这样后面接日志、监控、替换模型都会容易得多。

2. 核心概念拆解：Chain 与 Agent 的底层原理

2.1 自定义 Chain 的三种实现方式

LangChain 里自定义 Chain 目前常见的有三种，我按使用频率排个序：

第一种是直接用 Runnable 的组合能力。LangChain 核心的 Runnable 协议（对应invoke/batch/stream方法）非常方便，你写一个普通函数，用RunnableLambda包一下，再通过管道符|和 prompt、llm 串起来，就得到一条自定义链。这种方式代码量最省，适合节点较少的线性流水线。

from langchain_core.runnables import RunnableLambda, RunnableParallel from langchain_core.prompts import ChatPromptTemplate def clean_text(raw: str) -> str: # 这里可以做去重、去广告、规则清洗 return raw.strip() def split_sentences(text: str) -> list[str]: return [s for s in text.split("。") if s] prompt = ChatPromptTemplate.from_template("请总结以下内容：\n{text}") chain = ( RunnableLambda(clean_text) | RunnableLambda(split_sentences) | (lambda sentences: {"text": "。".join(sentences[:3])}) | prompt | llm )

第二种是继承Chain基类。这种方式更传统，适合需要显式定义输入输出 key、想让链条携带较多内部状态的场景。你需要实现input_keys、output_keys和_call方法，LangChain 会在调用前后校验 key，写起来略重，但结构最清晰。

from langchain.chains.base import Chain from typing import Any, Dict class ReportChain(Chain): llm: Any max_sentences: int = 3 @property def input_keys(self) -> set[str]: return {"source_text"} @property def output_keys(self) -> set[str]: return {"summary", "keyword"} def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]: source = inputs["source_text"] cleaned = self.clean(source) summary = self.llm.invoke(f"总结：{cleaned}") return {"summary": summary.content, "keyword": self.extract(cleaned)}

第三种是@chain装饰器，把普通函数直接标记成 Runnable。它介于前两者之间，函数本身就是一个链，适合把一个完整逻辑封装成可复用的节点。

这三种方式不存在谁一定更好，我的经验是：业务处理规则多的用 RunnableLambda 组合，因为函数单元最容易测试；要长期维护、需要严格 key 管理的项目用继承 Chain；临时封装一段逻辑给 Agent 工具复用，用@chain装饰器最顺手。

2.2 工具定义是 Agent 开发的地基

Agent 做的事情本质上是“决定调用哪个工具”，但真正干活的是工具本身。所以工具写得好不好，直接决定 Agent 的成败。

在 LangChain 中定义工具最标准的姿势是用@tool装饰器。你只需要注意一点：函数签名里的参数名、参数类型、以及 docstring，就是模型“认识”这个工具的全部信息来源。参数名要直观，docstring 要说明这个工具干什么、每个参数的含义、什么时候该用它。

from langchain_core.tools import tool @tool def get_sales_overview(date: str) -> str: """根据日期获取当日销售总额、订单量与退款金额，date 形如 2025-01-01。""" records = query_local_db(date) return f"销售总额:{records.total_amount}，订单量:{records.order_count}，退款:{records.refund_amount}"

这里有个细节容易被忽略：不要在一个工具里塞太多职责。比如我一开始写过一个analysis_and_suggest工具，既做统计又给建议，结果模型调用一次要传一堆参数，一旦某个参数传错就全废。后来我拆成get_sales_overview、compare_product_group、risk_alert三个工具，每个只做一件事，参数不超过三个，模型选工具的成功率大幅上升。

docstring 除了写“是什么”，还要写“什么时候用”。模型在做工具选择时会把用户问题和每个工具的 docstring 做语义匹配，你写得越贴近任务语言，它选得越准。比如对同一个销售查询接口，docstring 写“查询销售数据”和写“当需要了解某一天整体经营情况时使用，可返回总额、订单量、退款金额”，后者被选中的概率明显更高。

2.3 思考—行动—观察循环与 AgentExecutor

Agent 的核心运行机制是 ReAct 模式：思考（Reason）→ 行动（Act）→ 观察（Observation），循环往复直到得出最终答案。每一步模型输出一段文本，里面包含“思考”和“行动”指令，AgentExecutor 负责解析这些指令、调用对应工具、把结果放回上下文，让模型看到观察结果后继续思考。

我早期天真地以为 Agent 是框架内部自动做规划的，实际看过它的一步步日志才发现，AgentExecutor 只是一个很朴素的执行器：拿到模型输出，用正则或结构化解析提取 action 和 action_input，匹配工具列表，执行，把结果拼成 Observation 再送回去。真正做判断的始终是模型本身。

from langchain.agents import AgentExecutor, create_react_agent from langchain_core.prompts import PromptTemplate prompt = PromptTemplate.from_template(""" 你是一个运营分析助手，请尽可能使用工具回答问题。 工具列表如下： {tools} 工具说明： {tool_names} 请始终按如下格式回复： 思考：说明你打算做什么 行动：工具名称 行动输入：工具的 JSON 参数 观察：工具返回结果 （重复思考/行动/观察，直到得到结论） 最终回答：对用户的最终答复 用户问题：{input} """) agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) executor = AgentExecutor( agent=agent, tools=tools, max_iterations=5, handle_parsing_errors=True, verbose=True, )

这里的几个参数值得展开说说。max_iterations是安全阀，模型陷入循环时不会无限烧钱；handle_parsing_errors=True会让执行器在模型输出不符合格式要求时把错误信息退回给模型，让它修正自己；verbose=True方便开发期观察每一步思考过程。这些都是我在被实际坑过之后才加上的，后文会详细讲具体问题。

Agent 的循环机制我理解成一个“没有显式状态机的调度器”，它把调度决定权交给 LLM，同时靠工具列表和 prompt 约束它的行为边界。所以 Agent 工程的核心不是写循环，而是设计好工具集和提示词约束。

3. 实操落地：开发一个“每日运营分析助手”

3.1 场景定义与任务拆解

这一节我用一个模拟场景走完整流程。假设某电商业务每天会生成一份原始订单流水文本，文本里混着正常订单、活动补贴、退款记录和大量广告备注，运营希望每天能自动拿到一份结构化分析：整体经营概况、异常退款提醒、重点商品对比。

任务拆解后成两部分：

Chain 部分负责固定加工：清洗原始文本 → 抽取经营指标 → 生成摘要。这四步顺序不变、输入输出明确，用 Chain 固定下来。

Agent 部分负责动态分析：拿到 Chain 的摘要后，Agent 根据摘要内容决定是否调用销售总览工具、是否对比特定商品分组、是否发起退款异常提醒。这个决策完全不固定，某天一切正常可能一次工具都不用调，某天退款异常则需要连续查询多个工具。这种场景上 Agent 正合适。

为了代码贴近真实，我用模拟函数替代真实数据库，你替换成自己的存储层即可。整体流程这样设计也是基于一条实务考量：先让 Chain 把文本“煮熟”，Agent 就不用在噪声文本里折腾，token 更省、判断更准。

3.2 第一步：多阶段处理 Chain 的实现

我把 Chain 部分设计成三个节点：文本清洗、指标抽取、摘要生成。每个节点都写成普通函数，再用RunnableLambda串联。这样每个函数可以单独测试，中间任意一步出问题都能快速定位。

import json import re from langchain_core.runnables import RunnableLambda from langchain_core.prompts import ChatPromptTemplate # 节点1：清洗，去掉广告备注、多余空白、乱码字符 def clean_raw_text(raw: str) -> dict: lines = raw.split("\n") cleaned = [] for line in lines: if "广告" in line or "活动预告" in line: continue line = re.sub(r"\s+", " ", line).strip() if line: cleaned.append(line) return {"cleaned_text": "\n".join(cleaned)} # 节点2：结构化抽取，用 LLM 从清洗后的文本中抽字段 extract_prompt = ChatPromptTemplate.from_template(""" 从以下业务日志中抽取经营指标，以 JSON 格式返回： - order_count：订单总笔数 - total_amount：销售总额 - refund_count：退款笔数 - spu_groups：按商品分组统计的销量 日志内容： {cleaned_text} """) def parse_metrics(text: str) -> dict: # 假设 llm 返回的 content 是 JSON 字符串，这里做一次容错解析 try: return json.loads(text) except json.JSONDecodeError: # 兜底：截取花括号部分 start = text.find("{") end = text.rfind("}") + 1 return json.loads(text[start:end]) # 节点3：摘要，把结构化指标转成自然语言摘要 summary_prompt = ChatPromptTemplate.from_template(""" 你是一位运营分析师。请根据以下经营指标生成 3 句话摘要， 指出异常点（退款率高于 5% 时需要强调）。 指标： {metrics} """) def build_chain(llm): node_clean = RunnableLambda(clean_raw_text) node_extract = extract_prompt | llm | RunnableLambda(lambda x: parse_metrics(x.content)) node_summary = summary_prompt | llm chain = ( node_clean | RunnableLambda(lambda x: {"cleaned_text": x["cleaned_text"]}) | node_extract | RunnableLambda(lambda metrics: {"metrics": json.dumps(metrics, ensure_ascii=False)}) | node_summary | RunnableLambda(lambda x: {"final_summary": x.content, "metrics": x}) ) return chain

这里有几个设计细节我要强调。第一，清洗节点我用规则函数而不是 LLM，因为清洗是确定性的，用 LLM 又慢又不可控，正则足够。第二，抽取和摘要之间保留了 JSON 结构，摘要节点拿到的是结构化指标而不是原始文本，这样能用更少 token 达到同样的效果。第三，每个RunnableLambda转换返回的都是一个字典，因为下游 prompt 模板按 key 取参，统一字典格式能避免 key 错乱的坑。

实际开发中，我会在chain.invoke({"raw": 原始文本})之后打印中间节点结果，确保每一步的输出都符合预期，再继续连下一个节点。这一步不能省，尤其接入真实业务数据后，数据噪声带来的失败往往比模型本身的问题多得多。

3.3 第二步：多工具 Agent 的实现

Agent 部分我设计了三个工具：查询销售总览、对比商品分组、发起退款风险提醒。每个工具职责单一、参数精简，内部用模拟数据替换真实查询。

@tool def get_sales_overview(date: str) -> str: """根据日期获取该日整体经营指标，包括订单量、销售总额、退款金额。date 格式 YYYY-MM-DD。""" data = mock_query_overview(date) over = "" if data["refund_rate"] > 0.05: over = "，注意退款率已超过 5%" return f"订单量:{data['order_count']}，销售总额:{data['total_amount']}，退款金额:{data['refund_amount']}{over}" @tool def compare_product_group(date: str, group_ids: list[str]) -> str: """横向对比指定商品分组在某天的销量与销售额。group_ids 是商品分组 ID 列表。""" rows = [mock_query_group(date, gid) for gid in group_ids] text = "；".join([f"{r['group_id']}: 销量{r['sales_count']}, 销售额{r['amount']}" for r in rows]) return text @tool def risk_alert(date: str, threshold: float = 0.05) -> str: """当某天退款率超过 threshold 时，返回需要关注的订单类型分布。""" risk = mock_check_risk(date, threshold) if not risk: return "退款率正常，无需关注" return f"存在风险订单：{risk}"

构建 Agent 时，我把工具的列表传给create_react_agent，同时在 prompt 里强调“工具返回的数据要作为最终回答的依据，不要凭空编造”。这一步非常重要，模型拿到工具返回结果后仍然可能自行发挥，必须用 prompt 钳制它。

tools = [get_sales_overview, compare_product_group, risk_alert] prompt = PromptTemplate.from_template(""" 你是一名运营分析助手。你有以下工具可以使用： {tools} 工具名称列表： {tool_names} 请按如下步骤回答用户问题： 1. 思考用户需要什么信息 2. 选择最合适的工具，输出行动指令 3. 观察工具返回结果，若信息不足继续调用其他工具 4. 信息齐全后给出最终回答，最终回答必须基于工具结果，禁止编造 回复格式： 思考：... 行动：工具名 行动输入：JSON 参数 观察：... 最终回答：... 用户问题：{input} """) agent = create_react_agent(llm=llm, tools=tools, prompt=prompt) executor = AgentExecutor( agent=agent, tools=tools, max_iterations=4, handle_parsing_errors=True, verbose=True, )

我特别想提一下工具返回值的格式设计。工具返回的不应该是一串晦涩的 JSON，而是一段人类可读的自然语言，这样模型在下一步思考时可以直接引用。比如销售总览工具直接把“退款率已超过 5%”拼进结果，模型就能迅速意识到需要调用风险提醒工具。工具返回内容本身就是在替模型“思考前置条件”，这个巧劲在实战中非常管用。

3.4 第三步：把 Chain 和 Agent 串成一条完整工作流

Chain 和 Agent 各自就绪后，编排逻辑其实很简单：先跑 Chain 得到final_summary，把它作为用户问题的一部分传给 Agent。不要把原始文本传给 Agent，Agent 只需要基于摘要做决策就够了。

def run_pipeline(raw_text: str, date: str) -> str: # 1. Chain 加工 chain_result = chain.invoke({"raw": raw_text}) summary = chain_result["final_summary"] metrics = chain_result["metrics"] # 2. Agent 动态分析 user_question = f"今天是 {date}。运营摘要如下：{summary}\n请分析异常点，必要时使用工具获取更多数据。" result = executor.invoke({"input": user_question}) return result["output"]

这段串联逻辑看起来简单，但有几个隐藏细节：Chain 的结果要精简，因为 Agent 每轮都会把历史对话累积进上下文，你塞进去的内容越少，它能保持的注意力越集中；另外，日期字段要显式传入，不要让模型自己猜“今天”，避免时区或环境问题导致查询错位。

我实测过把完整原始文本和摘要都塞给 Agent 的情况，效果反而更差：模型被大量细节干扰，频繁调用无关工具。把信息压缩成“运营摘要”再给它，它的工具选择路径明显干净很多。

4. 常见问题与排查技巧实录

4.1 工具参数解析频频报错

使用 Agent 最常遇到的第一个坑就是模型输出的工具参数格式不对。比如工具定义参数是date: str，模型可能输出一个带中文“2025年1月1日”的值，或者干脆把多个参数合并成一句话。LangChain 在执行工具时一旦解析失败，整个 Agent 流程就会中断或进入重试。

我排查这个问题的顺序是：首先看工具定义里参数类型是否清晰，类型注解是模型生成参数的约束来源，尽量使用str、int、bool这类基础类型，不要用复杂泛型；其次看 docstring 有没有给出参数示例，比如“date 格式 YYYY-MM-DD”，模型看到示例后输出格式会稳定很多；最后给AgentExecutor打开handle_parsing_errors=True，让模型自己看到解析错误并修正。如果还不行，我会考虑用支持结构化输出的模型能力，把工具调用的输出格式强制约束成 JSON Schema。

另一个容易被忽视的点是工具 docstring 与实现不一致。模型根据 docstring 理解工具，如果你的 docstring 写的是“查询近七日数据”，但函数实现只查了某一天，模型就会按错误的理解使用。每次改动工具实现时，务必同步更新 docstring。

4.2 Agent 无限循环停不下来

Agent 陷入循环的表现是日志里反复出现同一个 action，模型不断调用同一个工具却得不到能回答用户问题的信息。常见原因是：工具返回内容与用户问题关联度不够，模型觉得“信息还不够”就再查一次；或者 prompt 里没有要求它“信息足够时必须停止”。

解决手段有三层。第一层是硬限制，max_iterations设置成 4 或 5，强制终止；第二层是软约束，在 prompt 里加一句“如果你认为现有工具结果已能回答问题，直接输出最终回答”；第三层是工具设计，让返回结果里包含明确的结论信号。比如风险提醒工具在“退款率正常”时返回“无需关注”，模型看到这个信号后就没有继续追问的动力。

我还试过把early_stopping_method设置为"generate"，当达到迭代上限时会强制模型生成一个最终回答而不是直接抛超时错误，这至少保证用户能拿到一个可用的结果。生产环境我建议同时开硬限制和生成兜底，避免流程空转。

4.3 自定义 Chain 拼接时 key 不匹配

用 Runnable 组合 Chain 时最常见的问题是某个节点返回的字典和下游 prompt 模板需要的变量名对不上。LangChain 本身对 key 的校验比较严格，一旦不匹配就直接报错，但这个报错信息往往要到运行期才暴露，调试起来很费劲。

我的习惯是给每个节点都固定返回统一的字典结构，并在初版开发时打印中间产物。你可以在这条链外面套一层调试函数：把chain.invoke的结果完整打印出来，对照上下游 key 是不是一致。另一个技巧是让所有节点的输出 key 都保持小写下划线风格，不要混用驼峰，减少人为失误。组合前先单独测试每个节点，确认输入输出正确再拼接，比组合后统一调试节省大量时间。

4.4 上下文过长导致效果跳水

Agent 多轮循环之后，所有历史思考和观察结果都会留在上下文里，几轮下来 token 消耗会快速上升，同时模型对早期信息的“注意力”明显下降。我遇到过第六轮工具调用时模型忘了最初用户问题的细节，导致回答偏题的情况。

解决方案是控制传入 Agent 的信息量。Chain 加工出来的摘要要压缩，工具返回的内容如果很长，在工具内部就要先提炼。必要时只在下一轮传最近一轮的观察结果，而不是把所有历史都塞进去。对于特别长的任务，可以用记忆模块做摘要历史，而不是逐字保留所有中间过程。

另外一个实用技巧是给工具返回值设定长度上限，超过一定长度自动截断并附上“……（已截断）”。模型处理精炼信息比处理一长串原始数字靠谱得多。

4.5 避坑速查表

这条表适合贴在项目文档里，每次接新场景时对照检查一遍，能省掉不少联调时间。

我个人在实际项目里最大的体会是：Agent 不是万能的，自定义 Chain 也不是炫技。复杂任务的正确解法永远是先靠 Chain 把确定性流程压得足够稳，再用 Agent 只负责那些真正需要临场判断的决策点。分开以后，链上任何一环出错都能定位到具体函数，Agent 选错工具也能通过调 docstring 或工具拆分来修正。如果你正准备改造自己的任务流，建议先从最小闭环开始：写一条三节点的 Chain 跑通数据加工，再挂一个只含两个工具的 Agent 验证决策路径，确认稳定后再逐步扩展工具集。实际跑起来你会发现，整个系统既可控，又有足够的智能空间。