大概每个把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。别急着上 AgentAgent 看起来更聪明但每次多一轮工具调用就多一层出错概率和延迟。更细致的判断可以参考下面这个表任务特征推荐组件原因步骤完全固定输入输出可预期Chain执行链路清晰调试容易流程固定但中间结果需要质量检查Chain 条件分支可在 Chain 节点之间插入校验逻辑需要从多个工具中选择调用Agent动态决策是 Agent 的核心能力工具调用顺序会随上下文变化Agent提前写死顺序会导致任务扩展性变差混合型先加工数据再动态分析Chain 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 思考—行动—观察循环与 AgentExecutorAgent 的核心运行机制是 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(llmllm, toolstools, promptprompt) executor AgentExecutor( agentagent, toolstools, max_iterations5, handle_parsing_errorsTrue, verboseTrue, )这里的几个参数值得展开说说。max_iterations是安全阀模型陷入循环时不会无限烧钱handle_parsing_errorsTrue会让执行器在模型输出不符合格式要求时把错误信息退回给模型让它修正自己verboseTrue方便开发期观察每一步思考过程。这些都是我在被实际坑过之后才加上的后文会详细讲具体问题。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_asciiFalse)}) | 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(llmllm, toolstools, promptprompt) executor AgentExecutor( agentagent, toolstools, max_iterations4, handle_parsing_errorsTrue, verboseTrue, )我特别想提一下工具返回值的格式设计。工具返回的不应该是一串晦涩的 JSON而是一段人类可读的自然语言这样模型在下一步思考时可以直接引用。比如销售总览工具直接把“退款率已超过 5%”拼进结果模型就能迅速意识到需要调用风险提醒工具。工具返回内容本身就是在替模型“思考前置条件”这个巧劲在实战中非常管用。3.4 第三步把 Chain 和 Agent 串成一条完整工作流Chain 和 Agent 各自就绪后编排逻辑其实很简单先跑 Chain 得到final_summary把它作为用户问题的一部分传给 Agent。不要把原始文本传给 AgentAgent 只需要基于摘要做决策就够了。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_errorsTrue让模型自己看到解析错误并修正。如果还不行我会考虑用支持结构化输出的模型能力把工具调用的输出格式强制约束成 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 避坑速查表常见现象根因解决方案工具参数解析失败类型注解或 docstring 不够清晰精简参数、写示例、开解析错误回传Agent 循环不退出工具结果缺乏结论信号、无迭代上限设 max_iterations、prompt 加终止规则Chain 拼接报 key 错误节点输出 key 与下游模板不一致统一命名、打印中间产物、单节点先行验证多轮后回答跑偏历史上下文过长稀释注意力压缩输入、只传必需信息、工具结果限长工具经常选错多个工具功能相似、docstring 含糊拆分职能、标注使用时机与场景LLM 直接回答不调工具prompt 未强制要求使用工具明示“必须基于工具结果回答”并给格式模板这条表适合贴在项目文档里每次接新场景时对照检查一遍能省掉不少联调时间。我个人在实际项目里最大的体会是Agent 不是万能的自定义 Chain 也不是炫技。复杂任务的正确解法永远是先靠 Chain 把确定性流程压得足够稳再用 Agent 只负责那些真正需要临场判断的决策点。分开以后链上任何一环出错都能定位到具体函数Agent 选错工具也能通过调 docstring 或工具拆分来修正。如果你正准备改造自己的任务流建议先从最小闭环开始写一条三节点的 Chain 跑通数据加工再挂一个只含两个工具的 Agent 验证决策路径确认稳定后再逐步扩展工具集。实际跑起来你会发现整个系统既可控又有足够的智能空间。