【AI应用开发】LangChain 框架有哪些缺点什么场景下会选择自研 Agent 调度目录LangChain 的核心价值LangChain 的十大缺点具体缺点深度拆解什么时候用 LangChain 就够什么时候必须自研自研 Agent 调度器设计对比总结1. LangChain 的核心价值先客观承认它的优点优点说明快速原型几行代码搭建 RAG/Agent demo生态丰富大量内置 Loader、Splitter、Retriever社区活跃问题容易找到解决方案标准化抽象Chain、Tool、Memory 等概念统一多模型支持一套代码切换 OpenAI/Anthropic/开源模型2. LangChain 的十大缺点1. 过度抽象 — 一行代码背后藏了 100 行逻辑 2. 调试困难 — 黑盒封装出问题不知从何查起 3. 性能开销 — 序列化/反序列化、回调链冗长 4. 版本不稳定 — API 频繁破坏性变更 (0.0.x → 0.1.x → 0.2.x) 5. 文档混乱 — 旧版 doc、新版 doc、迁移指南分散 6. 灵活性差 — 深层次定制需要 hack 源码 7. Prompt 管理弱 — 提示词散落各处难以维护 8. 流式支持差 — Streaming 层层拦截延迟高 9. 生产级缺失 — 缺少监控、熔断、重试等生产特性 10. 假简洁 — 简单场景代码少复杂场景比自研还多3. 具体缺点深度拆解3.1 过度抽象——“一行代码的代价”# LangChain 的 一行代码fromlangchain.agentsimportcreate_openai_functions_agent agentcreate_openai_functions_agent(llm,tools,prompt)# 这行代码背后发生的事:# 1. 创建 AgentExecutor# 2. 创建 OpenAIFunctionsAgent 实例# 3. 设置 output_parser (OpenAIFunctionsAgentOutputParser)# 4. 注入 callback handlers# 5. 创建 agent_scratchpad# 6. 序列化/反序列化中间消息格式# 7. ... 还有很多# 问题: 当 agent 行为不对时你根本不知道从哪个环节入手调试3.2 调试地狱# LangChain 调试体验fromlangchain.agentsimportAgentExecutor agent_executorAgentExecutor(agentagent,toolstools,verboseTrue)# verboseTrue 的输出: Entering new AgentExecutor chain... Action: search Action Input: {query: 退货流程} Observation: [一堆乱码般的长文本] Thought: ... Action: query_order ... [大量日志但关键信息被淹没] # 相比之下自研的调试体验classDebuggableAgent:defrun(self,user_input):traceTrace()withtrace.step(llm_reasoning):responseself.llm.chat(messages)trace.record(llm_input,messages[-3:])trace.record(llm_output,response)withtrace.step(tool_execution):forcallinresponse.tool_calls:resultself.execute_tool(call)trace.record(ftool:{call.name},{args:call.args,result:result,latency_ms:trace.step_latency()})# 清晰的、结构化的调试信息trace.print_tree()returnresponse# 输出: [LLM Reasoning] (1.2s) ├── Input: [user: 退货流程是什么?, system: 你是客服...] └── Output: calling tool search_knowledge_base [Tool: search_knowledge_base] (0.3s) ├── Args: {query: 退货流程} └── Result: 找到 3 条相关文档 (2345 chars) 3.3 版本不稳定真实经历: - LangChain 0.0.350 → 0.1.0: chain.run() 废弃, 改为 chain.invoke() - 0.1.x → 0.2.x: 大量 import 路径变化 - Agent 类型: ZeroShotAgent → OpenAIFunctionsAgent → 又变了 每次升级都是一次重构生产环境不敢轻易升级3.4 假简洁——复杂场景反而更复杂# LangChain 实现一个带记忆 多工具 条件终止的 Agent fromlangchain.memoryimportConversationBufferMemoryfromlangchain.agentsimportAgentExecutor,create_openai_functions_agentfromlangchain.toolsimporttoolfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain.callbacksimportStdOutCallbackHandlerfromlangchain.schemaimportAgentAction,AgentFinishfromlangchain.agents.output_parsersimportOpenAIFunctionsAgentOutputParser# ... 还要 20 行设置# 结果: 代码绕来绕去实际逻辑被淹没在框架样板代码中# 自研实现同样的功能 classMyAgent:def__init__(self,llm,tools,memory):self.llmllm self.toolstools self.memorymemorydefrun(self,user_input):messagesself.memory.get()[{role:user,content:user_input}]forstepinrange(15):responseself.llm.chat(messages,toolsself.tools)ifresponse.finish_reasonstop:self.memory.add(assistant,response.content)returnresponse.contentfortcinresponse.tool_calls:resultself._execute_tool(tc)messages.append({role:tool,content:str(result)})return处理超限def_execute_tool(self,tool_call):toolself.tools[tool_call.name]returntool.handler(**tool_call.args)4. 什么时候用 LangChain 就够✅ 快速原型和 MVP ✅ 简单的 RAG 应用文档问答 ✅ 工具少 5个的 Agent demo ✅ 个人项目或内部工具 ✅ 对框架有深度理解能从源码层面掌控5. 什么时候必须自研❌ 需要深度定制的 Agent 调度逻辑 ❌ 多 Agent 协作hierarchical/swarm ❌ 生产环境高并发 ( 100 QPS) ❌ 需要精确控制 Token 预算和循环终止 ❌ 需要在 Agent 循环中插入自定义逻辑 ❌ 需要精细的监控和可观测性 ❌ 团队没有 LangChain 专家——出了 bug 没人能修决策矩阵维度LangChain自研开发速度简单场景⭐⭐⭐⭐⭐⭐⭐⭐开发速度复杂场景⭐⭐⭐⭐⭐⭐可调试性⭐⭐⭐⭐⭐⭐⭐性能⭐⭐⭐⭐⭐⭐⭐⭐可维护性⭐⭐⭐⭐⭐⭐灵活性⭐⭐⭐⭐⭐⭐⭐学习曲线⭐⭐⭐⭐⭐⭐⭐⭐生产就绪⭐⭐⭐⭐⭐⭐⭐⭐6. 自研 Agent 调度器设计 自研 Agent 调度器核心设计 — 轻量化、可观测、可扩展 设计原则: 1. 代码量 500 行核心循环 2. 每个步骤可见、可记录、可回溯 3. 插入自定义逻辑只需加一个 hook 4. 状态机明确无隐式行为 fromdataclassesimportdataclass,fieldfromtypingimportAny,Callable,Optional,List,DictfromenumimportEnumimporttimeimportjsonimportlogging loggerlogging.getLogger(__name__)classAgentStatus(Enum):IDLEidleTHINKINGthinkingEXECUTINGexecutingWAITING_CONFIRMATIONwaiting_confirmationCOMPLETEDcompletedFAILEDfailedABORTEDaborteddataclassclassToolSpec:工具规格定义name:strdescription:str# 给 LLM 看的parameters:dict# JSON Schemahandler:Callable require_confirmation:boolFalsedataclassclassAgentStep:单步执行记录用于调试和回放step_num:intstatus:AgentStatus llm_input:Optional[List[Dict]]Nonellm_output:Optional[Any]Nonetool_calls:List[Dict]field(default_factorylist)latency_ms:float0token_usage:Dictfield(default_factorydict)dataclassclassAgentConfig:Agent 配置max_steps:int15max_retries:int3tool_timeout:float30.0llm_timeout:float20.0max_same_tool_calls:int2temperature:float0.1debug:boolFalseclassLightweightAgent:轻量级自研 Agent 调度器def__init__(self,config:AgentConfigNone):self.configconfigorAgentConfig()self.tools:Dict[str,ToolSpec]{}self.hooks:Dict[str,List[Callable]]{before_step:[],after_step:[],before_tool:[],after_tool:[],on_error:[],on_complete:[],}self.trace:List[AgentStep][]self.statusAgentStatus.IDLE self._circuit_breakerCircuitBreaker()# 工具注册 defregister_tool(self,tool:ToolSpec):self.tools[tool.name]tool logger.info(f工具已注册:{tool.name})defregister_tools(self,tools:List[ToolSpec]):fortintools:self.register_tool(t)# Hook 系统 defon(self,event:str,callback:Callable): 注册钩子 事件: before_step, after_step, before_tool, after_tool, on_error, on_complete ifeventinself.hooks:self.hooks[event].append(callback)returnself# 链式调用def_trigger(self,event:str,**kwargs):forcbinself.hooks.get(event,[]):try:cb(**kwargs)exceptExceptionase:logger.error(fHook{event}执行失败:{e})# 核心执行循环 asyncdefrun(self,user_input:str,messages:List[Dict]None,system_prompt:strNone,context:DictNone)-Dict: 主执行循环 Returns: {status: completed/failed/aborted, content: 回答内容, trace: [...]} # 初始化self.trace[]self.statusAgentStatus.THINKING# 构建初始消息ifmessagesisNone:messages[]ifsystem_prompt:messages.insert(0,{role:system,content:system_prompt})messages.append({role:user,content:user_input})# 主循环forstepinrange(1,self.config.max_steps1):step_recordAgentStep(step_numstep,statusAgentStatus.THINKING)step_starttime.time()# Before Hookself._trigger(before_step,stepstep,messagesmessages)try:# LLM 推理responseawaitself._call_llm(messages)step_record.llm_inputmessages[-5:]# 记录最后5条step_record.llm_outputresponse# 检查是否完成ifresponse.finish_reasonstop:self.statusAgentStatus.COMPLETED self._trigger(on_complete,responseresponse)returnself._format_result(completed,response.content,self.trace)# 执行工具调用ifresponse.tool_calls:self.statusAgentStatus.EXECUTINGfortcinresponse.tool_calls:# Before Tool Hookself._trigger(before_tool,tool_nametc.name,argstc.args)# 循环检测ifself._is_repeated_call(tc.name,tc.args):messages.append({role:system,content:该工具已用相同参数调用请不要重复。请基于已有结果回答。})continue# 执行工具resultawaitself._execute_tool(tc)step_record.tool_calls.append({name:tc.name,args:tc.args,result:result})# After Tool Hookself._trigger(after_tool,tool_nametc.name,argstc.args,resultresult)# 将工具结果加入消息messages.append({role:tool,tool_call_id:tc.id,content:json.dumps(result,ensure_asciiFalse)})# 终止条件检查should_stop,reasonself._check_termination(messages,step)ifshould_stop:breakexceptExceptionase:self._trigger(on_error,stepstep,errore)logger.error(f步骤{step}出错:{e})ifstepself.config.max_retries:self.statusAgentStatus.FAILEDreturnself._format_result(failed,str(e),self.trace)finally:step_record.latency_ms(time.time()-step_start)*1000self.trace.append(step_record)# After Step Hookself._trigger(after_step,step_recordstep_record)# 达到最大步数self.statusAgentStatus.ABORTEDreturnself._format_result(aborted,self._force_summarize(messages),self.trace)# 辅助方法 def_is_repeated_call(self,tool_name:str,args:dict)-bool:检测重复调用recent_calls[]forstep_recordinself.trace[-3:]:fortcinstep_record.tool_calls:iftc[name]tool_nameandtc[args]args:recent_calls.append(tc)returnlen(recent_calls)self.config.max_same_tool_callsdef_check_termination(self,messages,step):综合终止条件检查# 1. Token 预算total_tokenssum(len(str(m))forminmessages)//4iftotal_tokens12000:returnTrue,token_budget_exceededreturnFalse,asyncdef_call_llm(self,messages):调用 LLM带重试和超时# 实际实现根据使用的 LLM API 调整importasyncioreturnawaitasyncio.wait_for(self.llm.chat(messages,toolsself._format_tools()),timeoutself.config.llm_timeout)asyncdef_execute_tool(self,tool_call):执行工具带超时和熔断toolself.tools.get(tool_call.name)ifnottool:return{error:f工具 {tool_call.name} 不存在}ifnotself._circuit_breaker.allow(tool_call.name):return{error:工具已熔断请使用其他方式}importasynciotry:resultawaitasyncio.wait_for(asyncio.to_thread(tool.handler,**tool_call.args),timeoutself.config.tool_timeout)returnresultexceptasyncio.TimeoutError:self._circuit_breaker.record_failure(tool_call.name)return{error:f工具 {tool_call.name} 执行超时}exceptExceptionase:self._circuit_breaker.record_failure(tool_call.name)return{error:str(e)}def_format_tools(self):将工具格式化为 LLM API 需要的格式return[{type:function,function:{name:t.name,description:t.description,parameters:t.parameters}}fortinself.tools.values()]def_format_result(self,status,content,trace):return{status:status,content:content,trace:[{step:s.step_num,latency_ms:s.latency_ms,tool_calls:s.tool_calls}forsintrace]}def_force_summarize(self,messages):强制基于已有信息总结contextjson.dumps([{role:m[role],content:str(m.get(content,))[:200]}forminmessagesifm[role]in(tool,assistant)],ensure_asciiFalse)returnf处理步骤过多。基于已有信息总结如下:\n\n[上下文:{context}]classCircuitBreaker:简单的熔断器def__init__(self,threshold3,cooldown60):self.failures{}self.thresholdthreshold self.cooldowncooldowndefallow(self,tool_name):iftool_namenotinself.failures:returnTruecount,last_failself.failures[tool_name]ifcountself.threshold:returnTrueiftime.time()-last_failself.cooldown:self.failures.pop(tool_name)returnTruereturnFalsedefrecord_failure(self,tool_name):iftool_namenotinself.failures:self.failures[tool_name][0,0]self.failures[tool_name][0]1self.failures[tool_name][1]time.time()7. 对比总结LangChain自研如 LightweightAgent核心代码量依赖 50 模块~400 行理解成本需掌握框架抽象体系直接阅读循环逻辑调试verbose 模式 callbacktrace 结构化记录 hook定制需继承/覆写内部类直接改循环逻辑生产特性需额外集成内建熔断、重试、超时维护跟随框架更新完全自主Token 消耗框架 overhead ~200 tokens/轮几乎无 overhead最终建议原型/MVP 阶段用 LangChain 快速验证生产核心链路自研保证可控性和性能折中方案仅使用 LangChain 的 Loader/Splitter 等工具组件Agent 调度自研
【AI应用开发】LangChain 框架有哪些缺点什么场景下会选择自研 Agent 调度目录LangChain 的核心价值LangChain 的十大缺点具体缺点深度拆解什么时候用 LangChain 就够什么时候必须自研自研 Agent 调度器设计对比总结1. LangChain 的核心价值先客观承认它的优点优点说明快速原型几行代码搭建 RAG/Agent demo生态丰富大量内置 Loader、Splitter、Retriever社区活跃问题容易找到解决方案标准化抽象Chain、Tool、Memory 等概念统一多模型支持一套代码切换 OpenAI/Anthropic/开源模型2. LangChain 的十大缺点1. 过度抽象 — 一行代码背后藏了 100 行逻辑 2. 调试困难 — 黑盒封装出问题不知从何查起 3. 性能开销 — 序列化/反序列化、回调链冗长 4. 版本不稳定 — API 频繁破坏性变更 (0.0.x → 0.1.x → 0.2.x) 5. 文档混乱 — 旧版 doc、新版 doc、迁移指南分散 6. 灵活性差 — 深层次定制需要 hack 源码 7. Prompt 管理弱 — 提示词散落各处难以维护 8. 流式支持差 — Streaming 层层拦截延迟高 9. 生产级缺失 — 缺少监控、熔断、重试等生产特性 10. 假简洁 — 简单场景代码少复杂场景比自研还多3. 具体缺点深度拆解3.1 过度抽象——“一行代码的代价”# LangChain 的 一行代码fromlangchain.agentsimportcreate_openai_functions_agent agentcreate_openai_functions_agent(llm,tools,prompt)# 这行代码背后发生的事:# 1. 创建 AgentExecutor# 2. 创建 OpenAIFunctionsAgent 实例# 3. 设置 output_parser (OpenAIFunctionsAgentOutputParser)# 4. 注入 callback handlers# 5. 创建 agent_scratchpad# 6. 序列化/反序列化中间消息格式# 7. ... 还有很多# 问题: 当 agent 行为不对时你根本不知道从哪个环节入手调试3.2 调试地狱# LangChain 调试体验fromlangchain.agentsimportAgentExecutor agent_executorAgentExecutor(agentagent,toolstools,verboseTrue)# verboseTrue 的输出: Entering new AgentExecutor chain... Action: search Action Input: {query: 退货流程} Observation: [一堆乱码般的长文本] Thought: ... Action: query_order ... [大量日志但关键信息被淹没] # 相比之下自研的调试体验classDebuggableAgent:defrun(self,user_input):traceTrace()withtrace.step(llm_reasoning):responseself.llm.chat(messages)trace.record(llm_input,messages[-3:])trace.record(llm_output,response)withtrace.step(tool_execution):forcallinresponse.tool_calls:resultself.execute_tool(call)trace.record(ftool:{call.name},{args:call.args,result:result,latency_ms:trace.step_latency()})# 清晰的、结构化的调试信息trace.print_tree()returnresponse# 输出: [LLM Reasoning] (1.2s) ├── Input: [user: 退货流程是什么?, system: 你是客服...] └── Output: calling tool search_knowledge_base [Tool: search_knowledge_base] (0.3s) ├── Args: {query: 退货流程} └── Result: 找到 3 条相关文档 (2345 chars) 3.3 版本不稳定真实经历: - LangChain 0.0.350 → 0.1.0: chain.run() 废弃, 改为 chain.invoke() - 0.1.x → 0.2.x: 大量 import 路径变化 - Agent 类型: ZeroShotAgent → OpenAIFunctionsAgent → 又变了 每次升级都是一次重构生产环境不敢轻易升级3.4 假简洁——复杂场景反而更复杂# LangChain 实现一个带记忆 多工具 条件终止的 Agent fromlangchain.memoryimportConversationBufferMemoryfromlangchain.agentsimportAgentExecutor,create_openai_functions_agentfromlangchain.toolsimporttoolfromlangchain.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain.callbacksimportStdOutCallbackHandlerfromlangchain.schemaimportAgentAction,AgentFinishfromlangchain.agents.output_parsersimportOpenAIFunctionsAgentOutputParser# ... 还要 20 行设置# 结果: 代码绕来绕去实际逻辑被淹没在框架样板代码中# 自研实现同样的功能 classMyAgent:def__init__(self,llm,tools,memory):self.llmllm self.toolstools self.memorymemorydefrun(self,user_input):messagesself.memory.get()[{role:user,content:user_input}]forstepinrange(15):responseself.llm.chat(messages,toolsself.tools)ifresponse.finish_reasonstop:self.memory.add(assistant,response.content)returnresponse.contentfortcinresponse.tool_calls:resultself._execute_tool(tc)messages.append({role:tool,content:str(result)})return处理超限def_execute_tool(self,tool_call):toolself.tools[tool_call.name]returntool.handler(**tool_call.args)4. 什么时候用 LangChain 就够✅ 快速原型和 MVP ✅ 简单的 RAG 应用文档问答 ✅ 工具少 5个的 Agent demo ✅ 个人项目或内部工具 ✅ 对框架有深度理解能从源码层面掌控5. 什么时候必须自研❌ 需要深度定制的 Agent 调度逻辑 ❌ 多 Agent 协作hierarchical/swarm ❌ 生产环境高并发 ( 100 QPS) ❌ 需要精确控制 Token 预算和循环终止 ❌ 需要在 Agent 循环中插入自定义逻辑 ❌ 需要精细的监控和可观测性 ❌ 团队没有 LangChain 专家——出了 bug 没人能修决策矩阵维度LangChain自研开发速度简单场景⭐⭐⭐⭐⭐⭐⭐⭐开发速度复杂场景⭐⭐⭐⭐⭐⭐可调试性⭐⭐⭐⭐⭐⭐⭐性能⭐⭐⭐⭐⭐⭐⭐⭐可维护性⭐⭐⭐⭐⭐⭐灵活性⭐⭐⭐⭐⭐⭐⭐学习曲线⭐⭐⭐⭐⭐⭐⭐⭐生产就绪⭐⭐⭐⭐⭐⭐⭐⭐6. 自研 Agent 调度器设计 自研 Agent 调度器核心设计 — 轻量化、可观测、可扩展 设计原则: 1. 代码量 500 行核心循环 2. 每个步骤可见、可记录、可回溯 3. 插入自定义逻辑只需加一个 hook 4. 状态机明确无隐式行为 fromdataclassesimportdataclass,fieldfromtypingimportAny,Callable,Optional,List,DictfromenumimportEnumimporttimeimportjsonimportlogging loggerlogging.getLogger(__name__)classAgentStatus(Enum):IDLEidleTHINKINGthinkingEXECUTINGexecutingWAITING_CONFIRMATIONwaiting_confirmationCOMPLETEDcompletedFAILEDfailedABORTEDaborteddataclassclassToolSpec:工具规格定义name:strdescription:str# 给 LLM 看的parameters:dict# JSON Schemahandler:Callable require_confirmation:boolFalsedataclassclassAgentStep:单步执行记录用于调试和回放step_num:intstatus:AgentStatus llm_input:Optional[List[Dict]]Nonellm_output:Optional[Any]Nonetool_calls:List[Dict]field(default_factorylist)latency_ms:float0token_usage:Dictfield(default_factorydict)dataclassclassAgentConfig:Agent 配置max_steps:int15max_retries:int3tool_timeout:float30.0llm_timeout:float20.0max_same_tool_calls:int2temperature:float0.1debug:boolFalseclassLightweightAgent:轻量级自研 Agent 调度器def__init__(self,config:AgentConfigNone):self.configconfigorAgentConfig()self.tools:Dict[str,ToolSpec]{}self.hooks:Dict[str,List[Callable]]{before_step:[],after_step:[],before_tool:[],after_tool:[],on_error:[],on_complete:[],}self.trace:List[AgentStep][]self.statusAgentStatus.IDLE self._circuit_breakerCircuitBreaker()# 工具注册 defregister_tool(self,tool:ToolSpec):self.tools[tool.name]tool logger.info(f工具已注册:{tool.name})defregister_tools(self,tools:List[ToolSpec]):fortintools:self.register_tool(t)# Hook 系统 defon(self,event:str,callback:Callable): 注册钩子 事件: before_step, after_step, before_tool, after_tool, on_error, on_complete ifeventinself.hooks:self.hooks[event].append(callback)returnself# 链式调用def_trigger(self,event:str,**kwargs):forcbinself.hooks.get(event,[]):try:cb(**kwargs)exceptExceptionase:logger.error(fHook{event}执行失败:{e})# 核心执行循环 asyncdefrun(self,user_input:str,messages:List[Dict]None,system_prompt:strNone,context:DictNone)-Dict: 主执行循环 Returns: {status: completed/failed/aborted, content: 回答内容, trace: [...]} # 初始化self.trace[]self.statusAgentStatus.THINKING# 构建初始消息ifmessagesisNone:messages[]ifsystem_prompt:messages.insert(0,{role:system,content:system_prompt})messages.append({role:user,content:user_input})# 主循环forstepinrange(1,self.config.max_steps1):step_recordAgentStep(step_numstep,statusAgentStatus.THINKING)step_starttime.time()# Before Hookself._trigger(before_step,stepstep,messagesmessages)try:# LLM 推理responseawaitself._call_llm(messages)step_record.llm_inputmessages[-5:]# 记录最后5条step_record.llm_outputresponse# 检查是否完成ifresponse.finish_reasonstop:self.statusAgentStatus.COMPLETED self._trigger(on_complete,responseresponse)returnself._format_result(completed,response.content,self.trace)# 执行工具调用ifresponse.tool_calls:self.statusAgentStatus.EXECUTINGfortcinresponse.tool_calls:# Before Tool Hookself._trigger(before_tool,tool_nametc.name,argstc.args)# 循环检测ifself._is_repeated_call(tc.name,tc.args):messages.append({role:system,content:该工具已用相同参数调用请不要重复。请基于已有结果回答。})continue# 执行工具resultawaitself._execute_tool(tc)step_record.tool_calls.append({name:tc.name,args:tc.args,result:result})# After Tool Hookself._trigger(after_tool,tool_nametc.name,argstc.args,resultresult)# 将工具结果加入消息messages.append({role:tool,tool_call_id:tc.id,content:json.dumps(result,ensure_asciiFalse)})# 终止条件检查should_stop,reasonself._check_termination(messages,step)ifshould_stop:breakexceptExceptionase:self._trigger(on_error,stepstep,errore)logger.error(f步骤{step}出错:{e})ifstepself.config.max_retries:self.statusAgentStatus.FAILEDreturnself._format_result(failed,str(e),self.trace)finally:step_record.latency_ms(time.time()-step_start)*1000self.trace.append(step_record)# After Step Hookself._trigger(after_step,step_recordstep_record)# 达到最大步数self.statusAgentStatus.ABORTEDreturnself._format_result(aborted,self._force_summarize(messages),self.trace)# 辅助方法 def_is_repeated_call(self,tool_name:str,args:dict)-bool:检测重复调用recent_calls[]forstep_recordinself.trace[-3:]:fortcinstep_record.tool_calls:iftc[name]tool_nameandtc[args]args:recent_calls.append(tc)returnlen(recent_calls)self.config.max_same_tool_callsdef_check_termination(self,messages,step):综合终止条件检查# 1. Token 预算total_tokenssum(len(str(m))forminmessages)//4iftotal_tokens12000:returnTrue,token_budget_exceededreturnFalse,asyncdef_call_llm(self,messages):调用 LLM带重试和超时# 实际实现根据使用的 LLM API 调整importasyncioreturnawaitasyncio.wait_for(self.llm.chat(messages,toolsself._format_tools()),timeoutself.config.llm_timeout)asyncdef_execute_tool(self,tool_call):执行工具带超时和熔断toolself.tools.get(tool_call.name)ifnottool:return{error:f工具 {tool_call.name} 不存在}ifnotself._circuit_breaker.allow(tool_call.name):return{error:工具已熔断请使用其他方式}importasynciotry:resultawaitasyncio.wait_for(asyncio.to_thread(tool.handler,**tool_call.args),timeoutself.config.tool_timeout)returnresultexceptasyncio.TimeoutError:self._circuit_breaker.record_failure(tool_call.name)return{error:f工具 {tool_call.name} 执行超时}exceptExceptionase:self._circuit_breaker.record_failure(tool_call.name)return{error:str(e)}def_format_tools(self):将工具格式化为 LLM API 需要的格式return[{type:function,function:{name:t.name,description:t.description,parameters:t.parameters}}fortinself.tools.values()]def_format_result(self,status,content,trace):return{status:status,content:content,trace:[{step:s.step_num,latency_ms:s.latency_ms,tool_calls:s.tool_calls}forsintrace]}def_force_summarize(self,messages):强制基于已有信息总结contextjson.dumps([{role:m[role],content:str(m.get(content,))[:200]}forminmessagesifm[role]in(tool,assistant)],ensure_asciiFalse)returnf处理步骤过多。基于已有信息总结如下:\n\n[上下文:{context}]classCircuitBreaker:简单的熔断器def__init__(self,threshold3,cooldown60):self.failures{}self.thresholdthreshold self.cooldowncooldowndefallow(self,tool_name):iftool_namenotinself.failures:returnTruecount,last_failself.failures[tool_name]ifcountself.threshold:returnTrueiftime.time()-last_failself.cooldown:self.failures.pop(tool_name)returnTruereturnFalsedefrecord_failure(self,tool_name):iftool_namenotinself.failures:self.failures[tool_name][0,0]self.failures[tool_name][0]1self.failures[tool_name][1]time.time()7. 对比总结LangChain自研如 LightweightAgent核心代码量依赖 50 模块~400 行理解成本需掌握框架抽象体系直接阅读循环逻辑调试verbose 模式 callbacktrace 结构化记录 hook定制需继承/覆写内部类直接改循环逻辑生产特性需额外集成内建熔断、重试、超时维护跟随框架更新完全自主Token 消耗框架 overhead ~200 tokens/轮几乎无 overhead最终建议原型/MVP 阶段用 LangChain 快速验证生产核心链路自研保证可控性和性能折中方案仅使用 LangChain 的 Loader/Splitter 等工具组件Agent 调度自研
)
