把LangChain从“会用”推向“能真正处理复杂任务”核心就在于自己动手写Custom Chain和Custom Agent。内置的链解决线性流程还行一旦任务要求循环判断、多渠道取数、多工具协同、中途修改参数搭积木式的内置方案就完全不够用了。这篇文章我直接讲实际项目里沉淀下来的做法从Chain的类继承改造到Agent的ReAct循环控制再到一个多源数据整合案例的完整拆解重点会放在为什么这么设计以及哪些地方最容易炸。1. 为什么复杂任务必须走自定义Chain和Agent1.1 内置Chain的边界到底在哪里很多初学者觉得LangChain的内置Chain已经够多了LLMChain、RetrievalQAChain、ConversationChain好像串联几个就能应付日常开发。但真遇到复杂任务你会发现它们的核心假设是“流程固定且线性”。比如拿LLMChain做一次文本翻译、拿RetrievalQAChain做一次文档问答这些场景输入输出单一中间不会分叉内置实现完全够用。问题出在哪出在“流程”本身需要被动态决定。我举一个实际碰过的需求要写一个跨部门的月度报告系统需要先读销售数据、再比对库存、然后查历史异常记录最后根据前面的结果决定是生成正常报告还是触发告警。这个流程里有数据依赖有条件分支第二个步骤要不要执行完全取决于第一步的输出。这种结构如果用内置Chain硬拼你得把路由逻辑写在外层Python代码里链和链之间靠丑陋的全局变量传数据跑起来之后既难维护也难排错。所以第一课就是内置Chain适合那些你闭着眼睛都能画出流程图的场景一旦流程图里有菱形判断框就要考虑自己写Chain了。除了条件分支之外内置Chain还有一个隐藏缺陷——它对“状态”的支持太弱。复杂的任务通常意味着多轮工具调用每一轮的中间结果都可能影响下一轮的处理方式。比如“先搜索再根据搜索结果决定要不要追问”这种带状态流转的链路内置Chain的对象根本没有暴露一个可靠的状态容器你只能靠外部dict去hack代码写出来自己看着都难受。1.2 Agent相比Chain的差异和取舍Chain的本质是“人把流程定死模型只管执行”Agent的本质是“人给定目标和工具模型自己决定怎么走”。两相对比Agent最大的价值是应对不确定性比如用户问题本身描述不完整到底该查数据库还是查文档还是直接回答模型可以在运行时动态判断。但不要迷信Agent。我见过不少同行一上来就上个万能Agent结果发现三个问题token消耗直接翻倍、响应时延不可控、模型偶尔会乱调工具。Agent每走一步都要把当前状态所有中间结果再发给模型一次几轮循环下来上下文长度和费用都涨得很快。而且模型在做工具选择时出现幻觉是常态明明有更合适的工具它偏要用另一个。所以在真实项目里我的原则是能用Chain表达清楚的流程绝不上AgentAgent只用来处理那些“确实没办法提前定死”的分支环节。复杂系统最终往往是Chain加Agent的混合体外层用自定义Chain做状态机和流程控制内层挂Agent做动态决策点。这也是下面实战部分会用到的重要思路。2. 环境准备别在版本和依赖上浪费半天2.1 版本选择与依赖锁定LangChain迭代速度很快版本之间API经常不兼容。0.1版本里很常见的Chain.from_llm这类写法到0.2、0.3版本可能就挪到社区包甚至完全移除。这里强烈建议新项目直接用**langchain-core langchain 0.2**的组合核心抽象都收在langchain-core里通用编排能力在langchain里特定厂商的实现在langchain-community或langchain-{厂商名}里。python -m venv .venv source .venv/bin/activate pip install langchain0.2,0.4 langchain-core langchain-community langgraph装完之后立刻锁版本不要用飘忽不定的latestpip freeze requirements.txt为什么要强调锁版本因为LangChain生态里“今天能跑明天就报错”的情况实在太多。我记得有一次升级了一个patch版本结果langchain_core.runnables的导入路径变更CI直接挂了。做项目前把版本写死能省掉大量定位环境问题的时间。2.2 模型接入与基础配置模型接口在文章里我统一用get_llm()来表示实际项目中你只要替换成自己对接的模型客户端就行。比较合理的做法是写一个工厂函数把模型封装在后面这样后续换模型服务商不用动业务代码。# llm_factory.py from langchain_core.language_models import BaseChatModel def get_llm(model_name: str default) - BaseChatModel: # 这里返回一个你常用的ChatModel实例 # 可以用环境变量保存key不要硬编码到代码里 return your_chat_model_instance同时建议开启LangChain的详细日志输出测试阶段能看清楚每一步调用了什么import logging logging.basicConfig(levellogging.INFO)3. 自定义Chain从组合到类继承的必经之路3.1 LCEL组合式Chain的优雅和局限LCELLangChain Expression Language是这个框架里我最喜欢的设计用|把Runnable像管道一样串起来表达线性流程极其优雅from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_template(把下面内容翻译成英文{content}) chain prompt | get_llm() result chain.invoke({content: 你好世界})这看起来很美但LCEL有个隐蔽的问题它适合静态管道不适合动态逻辑。你可以用RunnableLambda往里面塞Python函数但函数之间的状态共享、分支选择都得靠外层代码兜着。等到链的数量超过三四个LCEL表达式会变得非常难读调试的时候分不清到底是哪个环节出了问题。这时候就要升级到“类继承”的自定义方式。本质上LangChain的所有Chain都是一个Runnable对象你只要实现invoke、batch、stream这几个核心方法就能完全控制执行逻辑。3.2 手写一个自定义Chain类先看一个最基本的自定义Chain骨架。它的目标不是实现什么复杂逻辑而是把模式讲清楚from typing import Any, Dict, List, Optional from langchain_core.runnables import RunnableConfig from langchain_core.runnables.base import Runnable class MyCustomChain(Runnable): def __init__(self, llm, retrieverNone): super().__init__() self.llm llm self.retriever retriever def invoke(self, inputs: Dict[str, Any], config: Optional[RunnableConfig] None) - Dict[str, Any]: # 1. 从inputs取出参数 question inputs.get(question, ) # 2. 做前置处理 if len(question.strip()) 0: return {answer: empty question, status: failed} # 3. 调用模型 resp self.llm.invoke(question) # 4. 整理输出结构 return { answer: resp.content, status: success, raw_response: resp }这段代码的核心点是它的输出不再只是模型返回的文本而是带状态、带过程信息的结构化内容。后面的流程拿status字段做判断拿raw_response做缓存都变得很自然。实现batch方法也很简单默认父类会循环调用invoke不过如果你想把一批请求并发发给模型就需要重写它def batch(self, inputs: List[Dict[str, Any]], config: Optional[RunnableConfig] None) - List[Dict[str, Any]]: from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers5) as executor: return list(executor.map(lambda x: self.invoke(x, config), inputs))3.3 多输入多输出与条件路由真正复杂的任务不会只用一条链所以自定义Chain更常见的形态是“路由 多分支”。举个例子我们做一个工单自动分类系统先判断工单类型咨询、故障、投诉再进入对应处理链。用自定义Chain实现时状态流转非常直观class TicketRouterChain(Runnable): def __init__(self, llm, chains: Dict[str, Runnable]): self.llm llm self.chains chains def invoke(self, inputs, configNone): content inputs.get(content, ) # 让模型输出一个类型标签 classification self.llm.invoke(f判断这个工单属于哪一类只输出一个词咨询/故障/投诉。工单内容{content}) category classification.content.strip() if category not in self.chains: return {category: category, result: unsupported category, status: failed} # 路由到具体处理链路 sub_result self.chains[category].invoke({content: content}) return {category: category, result: sub_result, status: success}这个Chain的价值不只是做了一个if-else路由而是把路由逻辑封装成了可复用、可跟踪的对象。你可以在外层记录每个工单走了哪条分支方便做数据分析。3.4 在自定义Chain里维护状态和记忆复杂任务的另一个典型需求是多轮对话中的状态管理。虽然LangChain有专门的MessageHistory但在自定义Chain中自己维护一个状态字典往往更可控。我常用的模式是在Chain内部维护一个self.state字典每次invoke时从inputs里读取增量更新然后存回去。class StatefulAssistantChain(Runnable): def __init__(self, llm): self.llm llm self.state {history: [], user_info: {}} def invoke(self, inputs, configNone): current_input inputs.get(input, ) user_id inputs.get(user_id, default) # 从外部传进来的历史兜底避免状态丢失 if history in inputs: self.state[history] inputs[history] history_text \n.join(self.state[history][-6:]) # 只保留最近6轮 prompt f你是客服助手。以下是历史对话\n{history_text}\n现在用户说{current_input}\n请回复。 resp self.llm.invoke(prompt).content self.state[history].append(f用户: {current_input}) self.state[history].append(f助手: {resp}) return {answer: resp, state: self.state}注意这里有个并发陷阱如果用同一个Chain实例处理多个用户请求self.state会被互相污染。简单做法是不要在全局复用实例或者根据user_id维护一个状态字典池。建议复杂的多用户场景不要自己硬写状态管理直接用外部的Redis或内存dict按sessionId隔离。Chain内只保存当前会话内的轻量状态。4. 自定义Agent让模型学会调用你的工具4.1 ReAct循环的本质LangChain里Agent的核心机制是ReAct循环一个不断重复“Thought - Action - Observation”的过程。模型先根据当前任务想一下该怎么做Thought然后决定调用哪个工具并给出参数Action系统工具执行后把结果返回Observation模型再根据观察结果思考下一步直到它认为已经能给出Final Answer为止。很多人在这一步栽跟头是因为没有把握住循环的终止条件。如果模型一直觉得观察结果不够就会无限循环下去所以Agent执行器必须同时具备两个终止条件一是模型主动输出Final Answer二是达到我们设定的最大迭代次数强行结束。LangChain里最简单的Agent其实可以直接交给AgentExecutor处理from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate tools [my_tool_1, my_tool_2] prompt PromptTemplate.from_template( 你可以使用以下工具遇到复杂问题请逐步思考。\n {tools}\n 工具名称列表{tool_names}\n 用户问题{input}\n 你的执行过程{agent_scratchpad} ) agent create_react_agent(llmget_llm(), toolstools, promptprompt) executor AgentExecutor( agentagent, toolstools, max_iterations5, return_intermediate_stepsTrue, handle_parsing_errorsTrue, )这段配置里有两个容易被忽略的关键参数。return_intermediate_stepsTrue会把每一步的Thought和Action完整记录下来后面排查“Agent为什么乱调工具”全靠它。handle_parsing_errorsTrue则是让模型在输出格式不对时自动纠错不至于整个Agent直接崩溃。4.2 如何写出高质量的自定义工具Agent的质量很大程度上取决于工具的描述质量。模型不是靠工具的函数名理解用途的而是靠description。我把一个常见的工具定义写出来from langchain_core.tools import tool tool def query_inventory(product_name: str) - str: 查询指定商品在当前仓库的实时库存返回可售数量和补货建议。参数是商品完整名称例如无线蓝牙耳机。 # 这里去调内部库存服务 result fake_inventory_api(product_name) return f商品[{product_name}]当前库存: {result[available]}件 tool def create_fulfillment_order(product_name: str, quantity: int, address: str) - str: 创建一条发货单需要商品名称、数量、收货地址。执行成功会返回发货单编号。 order_id fake_create_order(product_name, quantity, address) return f发货单已创建编号:{order_id}工具描述必须注意三点第一说清楚工具“能做什么”第二说清楚参数“应该填什么”最好给一个典型的参数示例第三说清楚返回结果长什么样因为模型要根据你的返回内容判断下一步。我踩过最痛的坑是工具描述太模糊。比如只写“查询库存”模型就经常把“商品名称”填成“商品型号”或者把数量参数也填进查询函数导致工具报错。后来在描述里加上“参数是商品的完整中文名称不是型号”调用准确率立刻从六成升到九成。4.3 自定义Agent执行逻辑不满足内置时自己写循环有时候AgentExecutor自带的循环控制满足不了需求比如我想在每轮Action之后做一次数据格式校验或者想把中间结果写入日志系统这时候手动实现ReAct循环更灵活def custom_agent_loop(prompt: str, tools, llm, max_steps: int 5): # 为Agent提供工具文案 tool_descs \n.join([f{t.name}: {t.description} for t in tools]) scratchpad [] current_input prompt for step in range(max_steps): # 让模型生成下一动作 response llm.invoke( f工具列表\n{tool_descs}\n f当前进度\n{scratchpad}\n f当前任务{current_input}\n 请输出下一步动作可以调用工具也可以直接给出最终答案。 ) text response.content # 判断是否结束 if FINAL_ANSWER: in text: return text.split(FINAL_ANSWER:)[-1].strip() # 解析工具名和参数 tool_name extract_tool_name(text) tool_args extract_tool_args(text) matched_tool next((t for t in tools if t.name tool_name), None) if matched_tool is None: scratchpad.append(f错误的工具 {tool_name}请重新选择) continue # 执行工具并捕获异常 try: observation matched_tool.invoke(tool_args) except Exception as e: observation f执行出错: {e} scratchpad.append(f调用 {tool_name}({tool_args}) - {observation}) return 达到最大执行步数强制结束这只是一个简化版真实项目里你还要处理“模型输出了非法JSON参数”的情况解析函数要做得足够健壮。但这段代码已经把Agent的本质撑起来了程序负责循环和容错模型负责决策两者协同。我在实际项目中很喜欢这种自定义写法因为它让每一步都可见可记录而且剥离了LangChain内置Agent大量的隐式prompt拼接出问题可以精准定位。4.4 Agent的容错与安全设计Agent一旦放开让模型自由调用工具风险就上来了。生产环境里绝对不能裸奔至少要做这四层保护工具限权Agent能调用的工具列表要最小化不要把你系统里所有API都暴露给模型。参数校验Agent传入的参数在真正执行前再做一次程序化校验比如数量必须是正整数地址不能为空。结果校验工具返回后增加一个轻量的verify环节防止模型拿着错误结果继续推导。成本看板必须记录每一步的token消费一旦单次任务超过预设预算就强制终止。下面这段是参数校验的示例def safe_invoke_tool(tool, args): # 二次校验参数 if quantity in args and not isinstance(args[quantity], int): try: args[quantity] int(float(args[quantity])) except (TypeError, ValueError): return 参数错误quantity必须是整数 if args.get(quantity, 0) 0: return 参数错误quantity必须大于0 return tool.invoke(args)5. 综合实战一个多源信息整合与自动处置系统5.1 场景设定与需求拆解这里我们做一个虚构但非常典型的场景内部运维团队收到一条系统告警需要自动判断告警等级查询相关服务状态再从知识库中搜索历史处理方案最后生成处置工单。任务看起来不算太复杂但对自动化的要求很高因为它有两个关键的不确定点第一告警信息可能不完整需要根据上下文去补全第二处理方案不能拍脑袋必须依据知识库和历史记录来决策。用固定Chain的话规则的维护成本太高用纯Agent的话又容易跑偏。最终的方案是用自定义Chain做整体流程的主干用Agent做中间“查资料和判断”的环节这样既有主干流程的确定性又有动态决策的灵活性。整个流程被拆成四步告警信息标准化服务状态查询调用工具历史方案检索调用工具处置方案生成与工单创建调用工具5.2 分层设计主控Chain与子Agent的配合我画了一个非常典型的分层结构不依赖任何图表库用文字描述最外层是一个AlertProcessingChain它接收告警原始文本输出处置工单数据。它内部维护一个step_state字典顺序执行四个步骤。第二步和第三步不是简单的工具调用而是交给一个内嵌Agent去完成因为这两步里“查什么”“怎么查”需要模型自行判断。主控Chain的代码骨架class AlertProcessingChain(Runnable): def __init__(self, llm, tools): self.llm llm self.tools tools self.agent_executor self._build_agent() def _build_agent(self): # 使用内置AgentExecutor限定工具范围 return AgentExecutor( agentcreate_react_agent( llmself.llm, toolsself.tools, promptPromptTemplate.from_template( 你是运维助手可以调用工具查询服务状态和历史处理方案。\n 工具列表{tools}\n 可以使用的工具名{tool_names}\n 用户任务{input}\n 执行记录{agent_scratchpad} ) ), toolsself.tools, max_iterations4, return_intermediate_stepsTrue, handle_parsing_errorsTrue, ) def invoke(self, inputs, configNone): raw_alert inputs.get(alert, ) state {raw: raw_alert, steps: {}} # 第一步标准化 state[steps][normalized] self._normalize_alert(raw_alert) # 第二步状态查询与历史检索交给Agent agent_result self.agent_executor.invoke({input: state[steps][normalized]}) state[steps][investigation] agent_result[output] # 第三步生成处置工单 state[steps][ticket] self._generate_ticket(state) return state def _normalize_alert(self, raw): resp self.llm.invoke(标准化以下告警信息提取出时间、服务名、异常码、影响范围\n raw) return resp.content def _generate_ticket(self, state): # 根据汇总信息生成结构化工单 return {title: 告警处置工单, detail: state[steps][investigation], priority: high}这里的关键点在于Agent只负责“调研”这一步而工单生成这种确定性逻辑仍然由Chain手写完成。Agent输出之后我们还保留了agent_result[intermediate_steps]可以用于后续审计看模型到底查了哪些工具。5.3 关键实现Agent与工具的接线为了让第二步的Agent真的会查服务状态我们要给它配两个工具tool def get_service_health(service_name: str) - str: 查询某个服务当前健康状态参数是服务英文名例如order-service。返回运行状态、错误率、最近重启时间。 health fake_health_check(service_name) return f{service_name} 状态{health[status]} 错误率{health[error_rate]} 最近重启{health[last_restart]} tool def search_incident_knowledge(keyword: str) - str: 搜索历史告警处理方案参数是描述性关键词例如memory leak或slow query。返回相关方案列表。 docs fake_search_kb(keyword) return \n.join([f[{d[title]}] {d[summary]} for d in docs])为了让Agent在第一步获取到更多的工具选择依据我们还在历史检索上加了点文章比如返回多个方案而不是一个让模型有对比空间。5.4 测试与效果评估这种多环节系统特别容易“每一步都对一点点合起来错很多”因此必须做分级测试。第一级单工具测试确认get_service_health在各类输入下都能正确返回第二级Agent单步测试人工构造几个告警文本看Agent是否选择了正确的工具第三级全链路测试准备一份20条的告警测试集把每条的工单结果人工打分。打分的重点不是“工单生成是否好看”而是三个核心指标能否正确判断是否要升级处理、能否在Agent循环中及时终止、能否在工具返回异常时不胡编一个处置结果。实测下来我发现Agent跑偏的情况多数发生在知识库返回内容很长的时候。模型会把不相关的历史方案也写进工单这时候两种解法一种是在工具返回结果里加一个“相关度”字段让模型优先看高相关度条目另一种是把Agent的prompt明确约束为“只引用与当前异常码知识库条目一致的内容”。后者效果更直接。6. 生产环境的坑调试、成本与稳定性6.1 高频报错与排查思路我把这段时间实际遇到过的Agent相关报错整理成一个速查表每个都附上了常用的排查方向。现象可能原因排查方法模型返回一堆废话不调用工具prompt里的工具描述太长模型注意力被稀释工具description精简控制在两三行内工具调用参数是幻觉值参数描述不够清楚模型不知道填什么在description中增加参数示例Agent无限循环max_iterations设置过大或工具返回结果让模型误判调低max_iterations同时检查工具返回文本解析失败却反复重试模型输出格式不稳定打开handle_parsing_errors或者把format instructions写得更死板工具执行超时拖垮整个Agent没有给工具设独立超时在工具内部用functools.wraps包一个超时机制其中工具执行超时是我踩得最狠的坑。有一次某个API偶发10秒高延迟Agent执行时等工具返回等得不耐烦模型判断“工具没反应”然后自动生成一个看似合理的假结果最后整条任务报警信息被污染。所以工具一定要设超时超时就返回一个“timeout”错误让模型知道这个结果不可靠。from functools import wraps import time def with_timeout(timeout_secs): def decorator(func): wraps(func) def wrapper(*args, **kwargs): result [None] def run(): result[0] func(*args, **kwargs) try: with ThreadPoolExecutor(max_workers1) as ex: future ex.submit(run) future.result(timeouttimeout_secs) except Exception as e: return f工具执行超时或异常: {e} return result[0] return wrapper return decorator6.2 成本控制与上下文管理自定义Agent是token消耗大户因为每一步Iteration都要把之前的工具返回重新塞给模型。有几个省钱技巧非常管用压缩工具返回工具返回内容如果特别长先做摘要再交给模型。比如知识库检索把返回的前3条做摘要而不是把全文塞进去。控制历史步数在自定义循环中scratchpad只保留最近2轮不要让模型看到太多旧的思考过程多数情况影响不大。模型分级工具选择和最终答案生成可以分别用不同档位的模型日常决策用便宜快速的模型最终汇总用更强的模型。下面的graph演示了如何对工具返回做压缩不是mermaid图是指代码结构def compress_observation(raw_obs: str, max_len: int 800) - str: if len(raw_obs) max_len: return raw_obs # 截断并补一句提示让模型自动忽略缺失部分 return raw_obs[:max_len] \n...内容过长已截断只能使用以上信息6.3 可观测性与审计Agent系统的黑盒属性比Chain严重得多。生产环境一定要记录三类日志决策日志Agent每一步的Thought内容也就是模型当时的推理过程工具日志每次工具调用名称、入参、出参、耗时会话日志完整prompt和最终输出方便事后复盘在自定义Chain里加一行记录很便宜self.history.append({ step: step, thought: text, tool: tool_name, args: tool_args, observation: observation, ts: time.time(), })如果用的是内置AgentExecutor别忘了设置return_intermediate_stepsTrue这个参数就是为了审计而生的。我见过不少团队上线前根本不知道这个参数出了问题只能猜Agent内部干了什么浪费了大量时间。做完整套系统之后我个人最深的感受是LangChain真正值得学的不是某个现成组件而是它提供的一套“把模型能力编排成流程”的思维框架。Chain和Agent不是二选一的竞争关系而是配合关系Chain负责把流程变成可以测试的代码Agent负责把决策变成可以迭代的模型调用。不管怎么设计一定要保证每个环节都能被记录、被还原、被复盘。这比任何技巧都重要。
