1. 项目概述LangChain工具系统的核心骨架如果你正在用LangChain构建一个智能体或者想把一个外部API、一个数据库查询、甚至一个简单的Python函数“教”给你的大模型让它能调用并完成复杂任务那么“工具”这个概念就是你绕不开的核心。LangChain的Tools系统本质上就是为大模型打造的“瑞士军刀”扩展包。它解决了一个根本问题大模型本身是“思考者”但它没有“手”和“眼”。工具就是它的手和眼让它能查询天气、发送邮件、执行计算从而从纯文本生成器进化为能执行实际任务的智能体。我最初接触时也被BaseTool、Callable、Runnable这些名词搞得有点晕感觉文档里每个字都认识连起来却不知道从哪下手。经过几个实际项目的折腾我才明白这三种定义方式并非互斥的选项而是代表了三种不同层级的抽象和灵活度分别适用于不同的开发阶段和场景。而串行与并行调用、错误处理与降级则是决定你的智能体是“玩具”还是“生产力工具”的关键工程细节。这篇文章我就以一个过来人的身份把这些核心概念掰开揉碎结合具体的代码示例和踩坑经验带你彻底掌握LangChain工具系统的设计哲学与实战技法。2. 核心类型深度解析BaseTool, Callable, Runnable理解这三种类型是构建健壮工具系统的第一步。它们不是简单的三种写法而是代表了LangChain对工具抽象的不同思考维度。2.1 BaseTool面向对象的标准化封装BaseTool是LangChain工具系统中最正式、功能最完整的基类。当你需要创建一个功能明确、需要良好描述、并且可能被多个智能体复用的工具时继承BaseTool是最佳选择。它的核心优势在于“标准化”。一个BaseTool的子类必须定义几个关键属性name: 工具的唯一标识符模型将用这个名字来调用它。description: 工具的详细描述。这是最重要的部分直接决定了模型是否、以及如何正确使用这个工具。描述要清晰说明工具的功能、输入格式和输出内容。_run或_arun方法工具的实际执行逻辑。一个典型的BaseTool定义示例from langchain.tools import BaseTool from pydantic import Field from typing import Optional, Type class WeatherQueryTool(BaseTool): 一个查询指定城市天气的工具。 name: str get_current_weather description: str ( 根据提供的城市名称查询该城市的当前天气情况。 输入应该是一个标准的城市名称字符串例如北京 或 New York。 输出将包含温度、天气状况和湿度。 ) city_name: str Field(..., description要查询天气的城市名称) def _run(self, city_name: str) - str: # 这里模拟一个天气API调用 # 真实场景中这里会是 requests.get(...) 调用某个天气服务 weather_data { 北京: 晴温度25°C湿度40%, 上海: 多云温度28°C湿度65%, New York: 小雨温度18°C湿度80%, } return weather_data.get(city_name, f未找到城市 {city_name} 的天气信息。) async def _arun(self, city_name: str) - str: # 异步版本适用于异步框架 # 这里为了简单直接调用同步方法实际应使用异步HTTP客户端 return self._run(city_name)实操心得与注意事项描述description是灵魂大模型如GPT完全依赖这个描述来决定何时调用工具。描述要尽可能精确、无歧义。我习惯采用“功能 输入格式 输出说明”的三段式写法效果很好。输入验证利用Pydantic的Field可以为工具参数添加更丰富的描述和验证规则这不仅能帮助模型理解也能在执行前捕获一些明显的错误。同步与异步如果你的工具涉及网络I/O如调用API强烈建议实现_arun异步方法这能显著提升在异步框架如FastAPI中智能体的整体吞吐量。如果只实现_run在异步链中调用可能会阻塞事件循环。2.2 Callable函数式编程的轻量级适配有时候你只是想快速把一个现有的Python函数包装成工具不想搞那么复杂的类定义。这时Callable方式就非常合适。它利用了Python的“鸭子类型”——任何实现了__call__方法的对象或者一个普通函数都可以被看作是一个Callable。LangChain提供了tool装饰器能极其方便地将一个普通函数转化为工具。使用tool装饰器的示例from langchain.tools import tool tool def search_wikipedia(query: str) - str: 在维基百科中搜索一个主题并返回摘要。输入应为一个搜索查询字符串。 # 这里简化处理真实情况会调用维基百科API print(f正在维基百科搜索: {query}) # 模拟返回 return f关于{query}的摘要这是一个模拟的搜索结果包含关键信息。 # 使用这个工具 search_tool search_wikipedia print(search_tool.name) # 输出search_wikipedia print(search_tool.description) # 输出在维基百科中搜索一个主题并返回摘要。输入应为一个搜索查询字符串。更灵活的手动包装方式from langchain.tools import Tool from typing import Callable def custom_calculator(expression: str) - str: 计算一个数学表达式。支持加减乘除例如3 5 * 2。 try: # 警告使用eval有安全风险仅作示例。生产环境应用ast.literal_eval或专用库。 result eval(expression) return f表达式 {expression} 的计算结果是{result} except Exception as e: return f计算表达式 {expression} 时出错{e} # 将函数包装成Tool实例 calc_tool Tool( nameCalculator, funccustom_calculator, description计算一个数学表达式字符串。例如输入 3 5 * 2 将返回 13。 )踩坑点函数签名与文档字符串tool装饰器会从函数的参数注解和文档字符串docstring中自动提取信息来生成name和description。因此为函数编写清晰、格式规范的文档字符串至关重要。如果文档字符串写得太随意模型可能无法正确理解工具的用途。Tool类包装当使用Tool类手动包装时你需要显式提供name,func,description三个参数。这种方式更灵活比如你可以包装一个类的方法或者一个lambda表达式。适用场景Callable方式非常适合原型验证、快速集成现有代码库或者工具逻辑非常简单的情况。它的缺点是功能不如BaseTool丰富例如缺少内置的参数验证需自己在函数内实现和更复杂的配置选项。2.3 Runnable融入LCEL执行流的一等公民Runnable是LangChain较新版本中提出的核心抽象旨在统一链Chain、工具Tool、模型Model等各类组件的接口。任何实现了.invoke()或.ainvoke()方法的对象都可以是一个Runnable。将工具定义为Runnable的最大好处是它可以无缝地嵌入到LangChain表达式语言LCEL中与其他Runnable组件如提示模板、模型、输出解析器通过管道符|优雅地组合。如何创建Runnable工具实际上前面提到的BaseTool和Tool实例本身都已经实现了Runnable接口。但我们可以更直接地利用LCEL来创建工具。from langchain_core.runnables import RunnableLambda # 1. 将一个普通函数包装成Runnable def fetch_user_profile(user_id: str) - str: 根据用户ID获取用户资料。输入应为用户ID字符串。 # 模拟数据库查询 profiles {001: 张三工程师擅长Python, 002: 李四设计师擅长UI} return profiles.get(user_id, 用户不存在) runnable_tool RunnableLambda(fetch_user_profile) # 现在它可以像其他Runnable一样使用 # result runnable_tool.invoke(001) # 2. 更复杂的Runnable工具可以组合其他步骤 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 假设我们有一个工具它先根据问题生成搜索关键词再调用搜索 prompt ChatPromptTemplate.from_template(请将以下问题转化为一个简洁的搜索关键词{question}) model ChatOpenAI(modelgpt-3.5-turbo) # 定义一个搜索函数模拟 def web_search(keyword: str) - str: return f搜索关键词 {keyword} 的结果模拟数据。 # 用LCEL组合成一个“搜索工具链”这个链本身就是一个Runnable工具 search_tool_chain prompt | model | (lambda x: x.content) | web_search # 这个 search_tool_chain 可以被智能体当作一个工具来调用核心优势与理解统一性在LCEL的世界里一切都是Runnable。工具作为Runnable可以和其他组件平等地组合、串联、并联极大地增强了代码的声明性和可读性。灵活性你可以轻松构建一个“工具链”其中包含多个步骤如问题清洗 - 模型思考 - 执行动作。这个链对外暴露的接口依然是一个简单的Runnable对智能体来说它就是一个功能更强大的“黑盒”工具。未来趋势LangChain正在大力推广LCEL将工具视为Runnable是更符合其设计哲学的方式。对于新建项目建议多考虑这种模式。三种方式的选择策略追求功能完整和复用性- 选BaseTool。追求快速原型和简单集成- 选Callable用tool装饰器。深度使用LCEL需要复杂组合逻辑- 将工具设计为或包装为Runnable。实际上它们互不排斥。一个BaseTool实例可以直接用在LCEL链中因为它本身就是Runnable。3. 工具调用模式串行与并行的工程抉择智能体如何使用多个工具是让它们一个接一个地执行串行还是同时发起多个请求并行这直接影响了任务的执行效率和逻辑正确性。3.1 串行调用顺序与依赖的经典模式串行调用是默认且最常见的模式。智能体根据模型对当前情况的理解一次只选择一个最合适的工具调用等待结果后再结合结果和上下文决定下一步动作继续调用工具或给出最终答案。这模仿了人类的顺序思考过程。实现方式在LangChain中当你将多个工具绑定到一个智能体如create_react_agent时智能体运行时本质上就是串行调用的。from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI # 假设我们已经有了两个工具weather_tool 和 calculator_tool tools [weather_tool, calculator_tool] # 拉取一个ReAct风格的提示模板 prompt hub.pull(hwchase17/react) # 创建智能体 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) agent create_react_agent(llm, tools, prompt) # 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 运行 - 这将触发串行调用 result agent_executor.invoke({ input: 北京现在的天气怎么样如果温度高于20度请计算这个温度华氏度是多少。 })执行过程可能如下模型思考“用户问了天气和计算。我需要先获取天气。”调用get_current_weather工具输入“北京”。得到结果“晴温度25°C湿度40%”。模型再次思考“我得到了温度25°C高于20度。现在需要计算25°C对应的华氏度。我需要计算器。”调用Calculator工具输入表达式25 * 9/5 32摄氏转华氏公式。得到结果“77”。模型综合信息给出最终答案。注意事项依赖处理串行模式天然适合有依赖关系的任务流如上例先有天气数据才能计算。思考开销每一步都需要模型进行“思考-行动-观察”的循环对于复杂任务调用次数可能较多总耗时和Token消耗会相应增加。错误累积前一步工具的失败或输出质量差会直接影响后续步骤。3.2 并行调用提升效率的激进策略并行调用是指智能体在一次“思考”中同时发起多个独立的工具调用。这适用于子任务间没有依赖关系且希望最大化节省总耗时的场景。重要提示LangChain的标准智能体框架如ReAct原生并不支持“一键式”并行调用。模型一次只输出一个Action。实现并行通常需要更定制化的策略。常见的并行化模式模型规划后并行执行让模型先制定一个计划列出所有可以并行执行的独立步骤然后由程序代码而非模型来并发执行这些工具。使用支持并行调用的特定智能体类型有些研究性或更高级的智能体框架设计了并行机制。手动拆分任务后并行在调用智能体前由开发者手动将一个大任务拆分成多个独立子任务然后使用asyncio.gather等并发工具来并行执行多个智能体实例每个处理一个子任务。示例模型规划后并行执行概念性代码import asyncio from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from typing import List # 定义规划步骤的数据结构 class ParallelPlan(BaseModel): steps: List[str] Field(description可以并行执行的独立任务列表) # 规划提示词 planning_prompt ChatPromptTemplate.from_template( 用户的问题是{question} 请分析这个问题并拆解出所有可以**同时独立执行**的子任务例如查询不同城市的天气、获取多个不同股票的价格。 请只列出这些子任务本身每个任务应该清晰到可以直接交给相应的工具执行。 输出格式必须是JSON{{steps: [任务1描述, 任务2描述, ...]}} ) # 假设我们有一个通用的“任务执行器”工具它能根据描述调用对应的真实工具 async def execute_task(task_description: str) - str: # 这里需要根据task_description的内容路由到具体的工具如weather_tool, calculator_tool # 这是一个复杂的路由逻辑此处简化处理 if 天气 in task_description: # 提取城市名调用天气工具... return await weather_tool._arun(北京) # 示例 elif 计算 in task_description: # 提取表达式调用计算器工具... return await calculator_tool._arun(35) else: return f无法处理任务{task_description} async def parallel_agent(question: str): llm ChatOpenAI(modelgpt-3.5-turbo) parser JsonOutputParser(pydantic_objectParallelPlan) planning_chain planning_prompt | llm | parser # 1. 规划阶段 plan await planning_chain.ainvoke({question: question}) print(f并行执行计划: {plan.steps}) # 2. 并行执行阶段 tasks [execute_task(step) for step in plan.steps] results await asyncio.gather(*tasks, return_exceptionsTrue) print(f并行执行结果: {results}) # 3. 结果汇总阶段可以再交给模型进行总结 # ... 汇总逻辑 ... # 使用 # asyncio.run(parallel_agent(查询北京和上海今天的天气并计算一下两地的平均温度是多少))并行调用的挑战与心得依赖识别自动、准确地识别任务间的独立性非常困难需要模型有很强的规划能力或者由开发者预设规则。结果整合并行执行得到的结果是分散的需要额外的逻辑通常是再调用一次模型来整合成连贯的最终答案。成本与复杂度并行可能减少总时间但可能增加总Token消耗因为需要额外的规划提示和代码复杂度。适用场景非常适合“信息收集”类任务比如同时查询多个数据源、批量获取用户信息等。对于强逻辑链的任务串行更安全。4. 错误处理与降级机制构建鲁棒的智能体任何外部工具调用都可能失败网络超时、API限流、输入无效、返回格式异常……一个健壮的智能体必须能妥善处理这些错误而不是直接崩溃或输出荒谬的结果。4.1 工具层面的错误处理首先应该在工具的内部实现中就包含基本的错误捕获。from langchain.tools import BaseTool import requests from tenacity import retry, stop_after_attempt, wait_exponential class RobustAPITool(BaseTool): name fetch_data description 从一个外部API获取数据。输入是API的URL。 retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def _run(self, url: str) - str: try: response requests.get(url, timeout10) response.raise_for_status() # 检查HTTP错误 return response.text except requests.exceptions.Timeout: return 错误请求超时。请检查网络或稍后重试。 except requests.exceptions.HTTPError as e: return f错误API请求失败状态码{e.response.status_code}。 except requests.exceptions.RequestException as e: return f错误网络请求异常 - {str(e)}。 except Exception as e: # 捕获其他未预期的异常 return f工具内部错误{str(e)}。关键点使用try...except包裹核心逻辑捕获预期内的异常超时、HTTP错误。友好的错误信息返回给模型的错误信息应该清晰、可读帮助模型理解发生了什么。避免直接抛出Python异常栈。重试机制对于瞬时的网络故障使用重试库如tenacity可以自动重试提高成功率。上例中我们配置了最多重试3次且等待时间指数增长。超时设置为所有网络请求设置合理的超时时间避免长时间阻塞。4.2 智能体执行器层面的错误处理AgentExecutor提供了处理工具运行时错误的机制。from langchain.agents import AgentExecutor agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue, # 处理模型输出无法解析为工具调用的错误 max_iterations5, # 防止无限循环 early_stopping_methodgenerate, # 当模型多次选择同一个无效工具时让其直接生成最终答案 ) # 即使工具出错执行器也会尝试继续 try: result agent_executor.invoke({input: 请用一个不存在的工具‘foo’做点什么。}) except ValueError as e: # 这里可能会捕获到解析错误或迭代超限错误 print(f代理执行失败: {e}) # 可以在这里实现降级逻辑例如让模型直接回答“我无法完成这个任务”AgentExecutor的关键参数handle_parsing_errors: 设为True后如果模型输出的内容无法被解析为有效的工具调用Action执行器会捕获这个错误并将错误信息反馈给模型让它“再试一次”。这非常有用。max_iterations和max_execution_time: 防止智能体陷入“思考-调用”的死循环是必须设置的安全阀。early_stopping_method: 当模型连续多次选择同一个无效工具时可以触发提前停止并让模型直接生成最终回复。4.3 降级策略当工具不可用时的备选方案降级Fallback是系统设计中保证可用性的重要手段。对于智能体来说降级意味着当首选工具失败时有备选方案。1. 工具级别的降级from langchain.tools import BaseTool class WeatherToolWithFallback(BaseTool): name get_weather description 获取城市天气优先使用A服务失败则尝试B服务。 def _run(self, city: str) - str: # 尝试主服务 result_from_a self._call_service_a(city) if result_from_a and 错误 not in result_from_a: return f[来自A服务] {result_from_a} # 主服务失败尝试降级服务 print(f服务A失败降级到服务B查询{city}) result_from_b self._call_service_b(city) if result_from_b: return f[来自B服务] {result_from_b} # 都失败 return f无法获取{city}的天气信息所有服务均不可用。 def _call_service_a(self, city): # 模拟调用 # if random.random() 0.7: # 模拟70%成功率 # return f{city}天气晴 # return 错误服务A超时 pass def _call_service_b(self, city): # 模拟调用 pass2. 智能体流程级别的降级当整个工具调用链失败时可以降级为让模型仅依靠自身知识来回答问题。from langchain.schema import AgentFinish from langchain.agents import AgentExecutor class RobustAgentExecutor(AgentExecutor): 一个自定义的、带降级功能的执行器 def _call(self, inputs): try: # 先尝试正常执行智能体流程 return super()._call(inputs) except Exception as e: # 如果执行过程中出现严重错误非工具解析错误 print(f智能体执行严重失败触发降级: {e}) # 降级逻辑直接让LLM基于原始输入和错误信息生成一个友好的回复 fallback_prompt f 用户的问题是{inputs[input]} 在尝试使用工具帮助你时系统遇到了意外错误{str(e)}。 请忽略工具仅依靠你自身的知识尽可能友好、有帮助地回答用户的问题。 你的回答 # 这里需要调用LLM假设我们有一个可用的llm对象 fallback_response self.agent.llm_chain.llm.invoke(fallback_prompt) return {output: fallback_response.content, intermediate_steps: []}3. 基于“Structured Output”的优雅错误返回对于Runnable工具可以定义结构化的输出其中包含状态码和错误信息字段让调用者能程序化地处理错误。from langchain_core.runnables import RunnableLambda from pydantic import BaseModel from typing import Optional class ToolResult(BaseModel): success: bool data: Optional[str] None error_message: Optional[str] None def robust_tool_func(input_str: str) - ToolResult: try: # ... 工具逻辑 ... result_data do_something(input_str) return ToolResult(successTrue, dataresult_data) except Exception as e: return ToolResult(successFalse, error_messagef工具执行失败: {str(e)}) runnable_tool RunnableLambda(robust_tool_func) # 调用者可以检查 result.success 来判断是否成功错误处理与降级的设计原则防御性编程工具内部要对输入进行验证对可能失败的操作进行异常捕获。可观测性错误信息要日志化便于调试。返回给用户或模型的错误信息应友好且不泄露内部细节。渐进式降级设计多级降级方案从重试、换备用工具到完全绕过工具。用户体验最终即使完全失败也应给用户一个清晰、得体的回应而不是一个技术性的崩溃信息。5. 实战构建一个具备错误处理和并行潜力的智能体让我们综合以上所有知识点构建一个相对完整的示例。这个智能体可以处理天气查询和计算并具备基本的错误处理和并行规划意识。import asyncio from typing import List, Optional from langchain.tools import BaseTool, Tool from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain import hub from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field import random # ---------- 1. 定义工具 (混合使用BaseTool和Callable) ---------- class WeatherTool(BaseTool): name get_weather description 查询指定城市的当前天气。输入是城市名称如‘北京’。 def _run(self, city: str) - str: # 模拟可能失败 if random.random() 0.2: # 20%概率模拟失败 raise ConnectionError(f模拟网络错误无法连接{city}的天气服务。) weather_map {北京: 25°C晴, 上海: 28°C多云, 伦敦: 15°C小雨} return weather_map.get(city, f未找到{city}的天气信息。) async def _arun(self, city: str) - str: return self._run(city) tool def calculator(expression: str) - str: 计算一个数学表达式例如 ‘(35)*2’。只支持基本算术。 try: # 安全警告生产环境请勿使用eval此处仅为演示。 allowed_chars set(0123456789-*/(). ) if not all(c in allowed_chars for c in expression): return 错误表达式中包含不安全字符。 result eval(expression) return str(result) except Exception as e: return f计算错误{str(e)} # ---------- 2. 创建带错误处理的工具包装器 ---------- def safe_tool_invoke(tool, tool_input): 安全调用工具捕获异常并返回统一格式。 try: if isinstance(tool_input, dict): result tool.invoke(tool_input) else: result tool.invoke({input: tool_input}) # 适配不同工具输入格式 return {success: True, result: result, tool_name: tool.name} except Exception as e: print(f工具 {tool.name} 调用失败: {e}) return {success: False, error: str(e), tool_name: tool.name} # ---------- 3. 并行规划器 (简化版) ---------- class ParallelTask(BaseModel): tasks: List[str] Field(description可以并行执行的独立任务描述列表) async def plan_parallel_tasks(question: str, llm) - List[str]: 让模型分析问题规划可并行任务。 prompt ChatPromptTemplate.from_template( 用户问题{question} 请判断为了解决这个问题是否可以并行执行多个独立操作 例如同时查询多个不同地点的天气、获取多个不同股票的价格等。 如果可以请列出这些可以并行执行的具体任务描述。 如果不可以或者你认为串行执行更合适请输出一个空列表 []。 只输出JSON格式{{tasks: [任务1, 任务2, ...]}} ) chain prompt | llm | JsonOutputParser(pydantic_objectParallelTask) try: plan await chain.ainvoke({question: question}) return plan.tasks except Exception: return [] # 规划失败退回串行 # ---------- 4. 主执行逻辑 ---------- async def main_agent(question: str): llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) weather_tool WeatherTool() calc_tool calculator # 注意tool装饰器返回的就是一个Tool实例 all_tools [weather_tool, calc_tool] # 第一步尝试并行规划 parallel_tasks await plan_parallel_tasks(question, llm) if parallel_tasks and len(parallel_tasks) 1: print(f检测到可并行任务: {parallel_tasks}) # 简化处理这里假设所有并行任务都调用天气工具实际需要更复杂的路由 # 使用异步并发执行 weather_tasks [] for task_desc in parallel_tasks: # 这里应该从task_desc中提取城市名简化处理直接使用任务描述 weather_tasks.append(safe_tool_invoke(weather_tool, task_desc)) # 注意safe_tool_invoke是同步函数真实异步环境应用异步版本并配合asyncio.gather print(f并行执行结果模拟: {weather_tasks}) # 将并行结果汇总作为上下文继续处理... combined_context .join([f{r[tool_name]}: {r.get(result, )} for r in weather_tasks if r[success]]) follow_up_question f基于以下并行查询结果{combined_context}。请回答原问题{question} question follow_up_question # 将汇总后的上下文作为新问题 # 第二步使用标准ReAct智能体处理可能是原问题也可能是汇总后的问题 prompt hub.pull(hwchase17/react) agent create_react_agent(llm, all_tools, prompt) agent_executor AgentExecutor( agentagent, toolsall_tools, verboseTrue, handle_parsing_errorsTrue, max_iterations6, early_stopping_methodgenerate ) try: final_result await agent_executor.ainvoke({input: question}) print(f\n最终答案: {final_result[output]}) except Exception as e: print(f\n智能体执行过程发生严重错误: {e}) # 最终降级直接让LLM回答 fallback_response await llm.ainvoke(f问题‘{question}’处理过程中遇到技术困难。请直接根据你的知识简要回答。) print(f降级回答: {fallback_response.content}) # ---------- 5. 运行示例 ---------- if __name__ __main__: # 测试一个可能触发并行规划的问题 # question 北京和上海现在的天气分别怎么样哪个更暖和 # 测试一个串行问题 question 如果北京气温25度那么华氏度是多少 asyncio.run(main_agent(question))这个示例集成了多种模式工具定义展示了BaseTool和tool装饰器两种方式。错误处理在WeatherTool中模拟了随机失败并通过safe_tool_invoke函数进行安全包装。并行探索通过一个独立的plan_parallel_tasks函数让模型先判断任务是否可并行并尝试规划。降级机制在main_agent的最后用try...except包裹了智能体执行器并在捕获到严重异常时直接降级调用LLM生成回答。构建生产级可用的智能体是一个系统工程LangChain提供的工具系统是强大的基石。理解BaseTool、Callable、Runnable这三种抽象能让你在合适的场景选择最优雅的实现方式。而妥善设计串行/并行调用策略并为其披上坚实的错误处理与降级铠甲则是让你的智能体从实验室走向真实世界的关键一步。在实际开发中往往需要根据具体的业务逻辑灵活混用这些模式和技术不断迭代和优化。