系列导航：[第一篇] → [第二篇] → [第三篇] → [第四篇] → [第五篇] → [第六篇] → 第七篇（当前） → 第八篇：Agent 开发 前置要求：已掌握前六篇内容，理解 LCEL 链的组合方式和 ChatModel 的调用机制 本篇目标：彻底掌握 LangChain 的工具调用体系。从"工具是什么"到"手动实现完整的工具调用循环"，覆盖工具定义的所有方式、如何把工具绑定给模型、Function Call 的底层机制、工具执行的错误处理，以及在不使用 Agent 的情况下自己控制调用流程。
系列导航：[第一篇] → [第二篇] → [第三篇] → [第四篇] → [第五篇] → [第六篇] → 第七篇（当前） → 第八篇：Agent 开发
前置要求：已掌握前六篇内容，理解 LCEL 链的组合方式和 ChatModel 的调用机制
本篇目标：彻底掌握 LangChain 的工具调用体系。从"工具是什么"到"手动实现完整的工具调用循环"，覆盖工具定义的所有方式、如何把工具绑定给模型、Function Call 的底层机制、工具执行的错误处理，以及在不使用 Agent 的情况下自己控制调用流程。
大模型本质上是一个文本处理器——它接收文本，输出文本。但很多真实任务需要模型与外部世界交互：查当前天气、执行数据库查询、调用第三方 API、搜索互联网……这些都不是纯文本推理能完成的。
工具调用（Tool Calling），也叫函数调用（Function Calling），解决的正是这个问题。它让模型在生成回答之前，能够先"决定"要调用哪个工具、传什么参数，然后根据工具的返回结果继续推理，最终给出答案。
这个过程用一个现实的类比来理解：想象你在电话里咨询一个顾问，顾问本身知识渊博，但他无法直接帮你查账单。他会对你说："你需要我查一下你的账单余额，我会先让助理去查，然后再告诉你结果。" 这个"决定要查账单、让助理去执行"的过程，就是工具调用的核心。
在技术层面，工具调用的完整流程是这样的：
第一步：开发者向模型描述工具的功能（工具名、参数、说明） 第二步：用户发起请求 第三步：模型判断是否需要工具，如果需要，输出"我要调用 X 工具，参数是 Y" （注意：模型本身不执行工具，只是"决策"） 第四步：你的代码接收这个决策，真正执行工具函数 第五步：把工具执行结果返回给模型 第六步：模型根据结果生成最终的自然语言回答
理解这个流程中"模型只决策、代码真执行"这一点非常关键，它直接影响你后面理解整个工具调用的架构设计。
LangChain 用 @tool 装饰器把普通的 Python 函数变成"工具"。被装饰后，函数会获得额外的元数据（名称、描述、参数 Schema），这些信息会被用来告诉模型"这个工具能做什么、怎么调用它"。
@tool
from langchain_core.tools import tool @tool def add_numbers(a: int, b: int) -> int: """计算两个整数的和并返回结果。""" return a + b
就这么简单。装饰完之后，add_numbers 就不只是一个普通函数了，我们来看看它变成了什么：
add_numbers
# 查看工具的基本属性 print(add_numbers.name) # add_numbers # （默认使用函数名，下划线风格） print(add_numbers.description) # 计算两个整数的和并返回结果。 # （来自函数的 docstring，这非常重要！） print(add_numbers.args_schema.schema()) # { # 'title': 'add_numbers', # 'type': 'object', # 'properties': { # 'a': {'title': 'A', 'type': 'integer'}, # 'b': {'title': 'B', 'type': 'integer'} # }, # 'required': ['a', 'b'] # } # （从函数签名的类型注解自动生成的 JSON Schema） # 工具仍然可以直接调用（作为普通函数使用） result = add_numbers.invoke({"a": 3, "b": 5}) print(result) # 8
@tool 装饰器实际上把函数信息自动转换成了 OpenAI Function Call 格式所需的 JSON Schema。这就是为什么类型注解和 docstring 在工具定义中如此重要——它们是模型理解工具的唯一依据。
工具的 docstring 是模型决定"要不要调用这个工具"的最重要依据。写得好的 docstring 能让模型在正确的时机调用正确的工具，写得差的 docstring 则会导致工具被误用或完全不被调用。
让我们看几个对比例子：
from langchain_core.tools import tool # ❌ 糟糕的工具描述：太模糊，模型无法判断何时使用 @tool def get_data(query: str) -> str: """获取数据。""" # 这个描述完全没有告诉模型：什么数据？从哪里获取？适用于什么场景？ pass # ❌ 也不好：描述了技术实现，但没说清楚使用场景 @tool def search_database(query: str) -> str: """在数据库中执行 SQL 查询。""" # 模型不应该知道底层用了 SQL，它只需要知道这个工具解决什么问题 pass # ✅ 良好的工具描述：明确说明适用场景、输入、输出 @tool def search_products(keyword: str) -> str: """ 根据关键词搜索商品信息。 当用户询问某种商品的价格、库存、规格或是否有货时使用此工具。 输入商品名称或关键词，返回匹配商品的列表，包含名称、价格和库存状态。 适用场景示例： - 用户问"有没有蓝牙耳机？" - 用户问"iPhone 15 多少钱？" - 用户问"查一下空调的库存" """ # 实际业务逻辑 return f"找到与'{keyword}'相关的商品：耳机A 299元 有货，耳机B 599元 无货"
一个好的工具描述通常包含三个要素：这个工具做什么（一句话概括）、什么时候使用它（使用场景）、输入输出是什么（参数和返回值的语义）。
对于参数较多或参数含义不明显的工具，可以用 Pydantic 的 Field 为每个参数添加描述：
Field
from langchain_core.tools import tool from pydantic import Field from typing import Optional @tool def book_hotel( city: str = Field(description="目标城市名称，例如：北京、上海、成都"), check_in_date: str = Field(description="入住日期，格式：YYYY-MM-DD，例如 2025-06-01"), check_out_date: str = Field(description="退房日期，格式：YYYY-MM-DD，例如 2025-06-03"), guests: int = Field(default=1, description="入住人数，默认 1 人"), room_type: Optional[str] = Field( default=None, description="房间类型，可选值：'standard'（标准间）、'deluxe'（豪华间）、'suite'（套房）。不指定则显示所有类型" ), ) -> str: """ 查询并预订酒店房间。 当用户需要预订酒店或查询酒店房间时使用此工具。 会根据城市、日期和人数返回可用房间信息和预订确认。 """ nights = 1 # 简化计算 return ( f"已在{city}为{guests}位客人预订 {check_in_date} 至 {check_out_date} 的酒店。" f"房型：{room_type or '标准间'}，共 {nights} 晚。" ) # 查看参数 schema，可以看到每个参数都有了清晰的描述 import json print(json.dumps(book_hotel.args_schema.schema(), ensure_ascii=False, indent=2))
工具的返回值会被原封不动地发回给模型，模型需要理解这个返回值并基于它生成回答。返回值的设计直接影响模型的理解质量：
from langchain_core.tools import tool from datetime import datetime import json # ❌ 返回值难以理解的工具 @tool def get_weather_bad(city: str) -> str: """获取天气。""" return "25 1013 60 NW 3" # 模型不知道这些数字代表什么 # ✅ 返回值清晰的工具 @tool def get_weather(city: str) -> str: """ 获取指定城市的实时天气信息。 返回包含温度、天气状况、湿度和风向的格式化文本。 当用户询问天气、出行是否需要带伞、穿什么衣服时使用此工具。 """ # 实际项目中会调用天气 API，这里用模拟数据 weather_data = { "city": city, "temperature": "25°C", "feels_like": "27°C", "condition": "多云转晴", "humidity": "60%", "wind": "西北风 3 级", "uv_index": "中等", "suggestion": "天气舒适，适合出行" } # 返回结构化的自然语言描述，模型更容易理解 return ( f"{city}当前天气：{weather_data['condition']}，" f"气温 {weather_data['temperature']}（体感 {weather_data['feels_like']}），" f"湿度 {weather_data['humidity']}，{weather_data['wind']}。" f"建议：{weather_data['suggestion']}" ) # 注意：返回值类型不只是 str，也可以是任何可序列化的类型 @tool def get_stock_price(ticker: str) -> dict: """ 获取股票实时价格和涨跌幅。 参数 ticker 是股票代码，例如 AAPL（苹果）、TSLA（特斯拉）、600519（贵州茅台）。 """ # 模拟数据 return { "ticker": ticker, "current_price": 185.50, "change": +2.30, "change_percent": "+1.26%", "market_cap": "2.89T", "timestamp": datetime.now().isoformat(), }
@tool 装饰器虽然简洁，但在某些场景下不够灵活——比如工具的 Schema 需要在运行时动态生成，或者你想把现有的类方法包装成工具。这时候可以用 StructuredTool 类：
StructuredTool
from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field from typing import Optional # 首先定义输入参数的 Pydantic Schema class DatabaseQueryInput(BaseModel): """数据库查询工具的输入参数""" table: str = Field(description="要查询的表名，例如：users、orders、products") conditions: Optional[str] = Field( default=None, description="查询条件（WHERE 子句内容），例如：age > 18 AND city = '北京'" ) limit: int = Field( default=10, description="返回记录数量上限，默认 10 条，最大 100 条", ge=1, le=100, # 使用 Pydantic 验证：1 <= limit <= 100 ) order_by: Optional[str] = Field( default=None, description="排序字段，例如：created_at DESC" ) # 实际的查询函数 def execute_database_query( table: str, conditions: Optional[str] = None, limit: int = 10, order_by: Optional[str] = None, ) -> str: """真正执行查询的函数（这里用模拟数据代替真实数据库操作）""" where_clause = f" WHERE {conditions}" if conditions else "" order_clause = f" ORDER BY {order_by}" if order_by else "" sql = f"SELECT * FROM {table}{where_clause}{order_clause} LIMIT {limit}" # 模拟返回结果 return f"执行查询：{sql}\n结果：找到 3 条记录（模拟数据）" # 用 StructuredTool 包装 db_query_tool = StructuredTool.from_function( func=execute_database_query, name="query_database", # 自定义工具名（不用函数名） description="""查询业务数据库中的数据。 当用户需要查看用户信息、订单详情、产品库存等业务数据时使用此工具。 支持条件过滤、排序和数量限制。 注意： - 只能查询，不能修改数据 - 表名必须是已存在的业务表 - 条件语句使用 SQL WHERE 子句格式""", args_schema=DatabaseQueryInput, # 使用自定义的 Pydantic Schema ) print(db_query_tool.name) # query_database print(db_query_tool.description) # 上面写的描述 # 调用工具（会经过 Pydantic 验证） result = db_query_tool.invoke({ "table": "orders", "conditions": "status = 'pending'", "limit": 5, "order_by": "created_at DESC", }) print(result)
在面向对象的项目中，业务逻辑通常封装在类里。StructuredTool 可以把类方法包装成工具：
from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field from typing import List, Optional class EmailService: """邮件服务类（模拟真实的邮件客户端）""" def __init__(self, sender_email: str, smtp_config: dict): self.sender = sender_email self.config = smtp_config self.sent_emails = [] # 记录已发送的邮件 def send_email(self, to: str, subject: str, body: str, cc: Optional[str] = None) -> str: """发送邮件""" email_record = { "from": self.sender, "to": to, "cc": cc, "subject": subject, "body_preview": body[:50], } self.sent_emails.append(email_record) return f"邮件已成功发送给 {to}，主题：{subject}" def search_emails(self, keyword: str, max_results: int = 5) -> str: """搜索邮件""" # 模拟搜索结果 return f"找到 {max_results} 封包含 '{keyword}' 的邮件（模拟数据）" class SendEmailInput(BaseModel): to: str = Field(description="收件人邮箱地址") subject: str = Field(description="邮件主题") body: str = Field(description="邮件正文内容") cc: Optional[str] = Field(default=None, description="抄送邮箱地址，不需要则省略") class SearchEmailInput(BaseModel): keyword: str = Field(description="搜索关键词，在邮件主题和正文中搜索") max_results: int = Field(default=5, description="返回结果数量，默认 5 封") # 实例化服务 email_service = EmailService( sender_email="assistant@company.com", smtp_config={"host": "smtp.company.com", "port": 587} ) # 把实例方法包装成工具（通过 lambda 绑定 self） send_email_tool = StructuredTool.from_function( func=lambda **kwargs: email_service.send_email(**kwargs), name="send_email", description="发送邮件给指定收件人。当用户需要发送邮件、回复邮件或转发邮件时使用。", args_schema=SendEmailInput, ) search_email_tool = StructuredTool.from_function( func=lambda **kwargs: email_service.search_emails(**kwargs), name="search_emails", description="在邮件箱中搜索邮件。当用户需要查找特定邮件、历史邮件时使用。", args_schema=SearchEmailInput, ) # 测试 print(send_email_tool.invoke({"to": "boss@company.com", "subject": "项目进展", "body": "项目进展顺利..."})) print(search_email_tool.invoke({"keyword": "项目", "max_results": 3}))
定义好工具之后，需要用 bind_tools() 把工具"注册"给模型。这一步在底层会把工具的 Schema 转换成模型 API 支持的格式（OpenAI Function Call 格式）并附加到请求中。
bind_tools()
from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def get_current_time(timezone: str = "Asia/Shanghai") -> str: """ 获取指定时区的当前时间。 当用户询问现在几点、当前时间等时使用。 timezone 使用 IANA 时区格式，默认为北京时间（Asia/Shanghai）。 """ from datetime import datetime import pytz tz = pytz.timezone(timezone) now = datetime.now(tz) return now.strftime(f"%Y年%m月%d日 %H:%M:%S（{timezone}）") @tool def calculate(expression: str) -> str: """ 计算数学表达式的结果。 支持基本四则运算和常见数学函数。当用户需要做数学计算时使用此工具。 输入标准的数学表达式，例如：2 + 3 * 4, sqrt(16), 100 / 7 """ import math try: # 安全地执行数学表达式（生产环境需要更严格的沙箱） allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("_")} result = eval(expression, {"__builtins__": {}}, allowed_names) return f"{expression} = {result}" except Exception as e: return f"计算失败：{e}" # 创建工具列表 tools = [get_current_time, calculate] # 初始化模型 llm = ChatOpenAI(model="gpt-4o", temperature=0) # 将工具绑定到模型 # bind_tools 返回一个新的 Runnable，不修改原始 llm 对象 llm_with_tools = llm.bind_tools(tools) # 现在来看看模型对一个需要工具的问题是如何响应的 response = llm_with_tools.invoke("现在北京时间几点？") print("回复内容：", response.content) # 通常是空字符串，因为模型决定要用工具而不是直接回答 # 内容："" print("工具调用：", response.tool_calls) # [{'name': 'get_current_time', 'args': {'timezone': 'Asia/Shanghai'}, 'id': 'call_abc123'}] # 模型告诉我们它要调用 get_current_time，参数是 Asia/Shanghai
这里有个关键的概念需要理解：当模型决定使用工具时，它的 response.content 通常是空的，真正重要的信息在 response.tool_calls 里。tool_calls 是一个列表，每个元素包含： - name：要调用的工具名称 - args：工具的参数（字典格式） - id：这次工具调用的唯一 ID（用于后续把结果对应回去）
response.content
response.tool_calls
tool_calls
name
args
id
绑定工具后，模型的响应（AIMessage）有了更丰富的结构，值得深入了解：
AIMessage
from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def search_web(query: str) -> str: """搜索互联网上的最新信息。当用户询问最新新闻、实时数据或需要查询不确定信息时使用。""" return f"搜索'{query}'的结果：（模拟搜索结果）" @tool def get_weather(city: str) -> str: """获取城市天气。""" return f"{city}：晴天，25°C" llm_with_tools = ChatOpenAI(model="gpt-4o").bind_tools([search_web, get_weather]) # 提一个需要多个工具的问题 response = llm_with_tools.invoke("帮我查一下上海今天的天气，以及今天的科技新闻头条") print("=" * 50) print(f"消息类型：{type(response).__name__}") print(f"content（直接回复）：'{response.content}'") print(f"tool_calls 数量：{len(response.tool_calls)}") print() for i, call in enumerate(response.tool_calls): print(f"工具调用 {i+1}：") print(f" 工具名：{call['name']}") print(f" 参数：{call['args']}") print(f" 调用ID：{call['id']}") print() # 输出示例： # 消息类型：AIMessage # content（直接回复）：'' # tool_calls 数量：2 # # 工具调用 1： # 工具名：get_weather # 参数：{'city': '上海'} # 调用ID：call_Abc123 # # 工具调用 2： # 工具名：search_web # 参数：{'query': '今日科技新闻头条'} # 调用ID：call_Def456
模型可以同时决定调用多个工具（如上例），这说明它判断这两个问题需要并行获取信息。在实际实现中，我们可以并行执行这些工具以节省时间。
有时候你希望强制模型使用某个特定工具，或者限制它的选择范围：
from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def search(query: str) -> str: """搜索信息。""" return f"搜索结果：{query}" @tool def calculate(expr: str) -> str: """计算。""" return f"计算结果：{expr}" llm = ChatOpenAI(model="gpt-4o") # 默认：让模型自己决定是否调用工具（auto） llm_auto = llm.bind_tools([search, calculate], tool_choice="auto") # 强制必须调用某个工具（指定工具名） llm_force_search = llm.bind_tools([search, calculate], tool_choice="search") # 强制必须调用工具（但让模型自己选哪个） llm_must_use_tool = llm.bind_tools([search, calculate], tool_choice="required") # 禁止调用任何工具（纯文本回答） llm_no_tools = llm.bind_tools([search, calculate], tool_choice="none") # 强制调用的场景：比如你做了一个"信息提取"应用， # 输入是原始文本，输出必须经过提取工具格式化 response = llm_force_search.invoke("告诉我关于 LangChain 的信息") print(response.tool_calls) # 无论用户问什么，都会触发 search 工具
很多教程直接跳到 Agent，但其实理解"手动控制工具调用循环"对于深入掌握 LangChain 工具体系非常重要。手动控制意味着你自己写循环逻辑，决定什么时候执行工具、什么时候结束。
这种方式在以下场景中很有价值： - 你需要在工具执行前后加入自定义逻辑（比如权限检查、日志记录） - 你需要精确控制工具调用的次数或条件 - 你想理解 Agent 的底层工作原理（Agent 本质上就是自动化了下面这个循环）
from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage # 定义工具 @tool def get_weather(city: str) -> str: """获取指定城市的当前天气。""" # 模拟天气 API weather_db = { "北京": "晴天，气温 22°C，西北风 3 级，空气质量良", "上海": "多云，气温 26°C，东南风 2 级，湿度 75%", "广州": "阵雨，气温 28°C，南风 4 级，注意防雨", "成都": "阴天，气温 18°C，无风，空气质量优", } return weather_db.get(city, f"暂无 {city} 的天气数据") tools = [get_weather] # 创建工具查找字典，方便后面根据名字找到对应工具 tool_map = {t.name: t for t in tools} llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) # -------- 第一步：发送用户问题 -------- user_question = "北京和上海今天天气怎么样？" messages = [HumanMessage(content=user_question)] print(f"用户问：{user_question}\n") # 发送给模型，得到"要调用哪些工具"的决策 ai_response = llm_with_tools.invoke(messages) messages.append(ai_response) # 把模型的决策加入消息历史 print(f"模型决策：需要调用 {len(ai_response.tool_calls)} 个工具") for call in ai_response.tool_calls: print(f" → 调用 {call['name']}，参数：{call['args']}") print() # -------- 第二步：执行工具 -------- # 遍历模型要求调用的所有工具，逐一执行 for tool_call in ai_response.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] tool_id = tool_call["id"] # 每次工具调用都有唯一 ID # 找到对应的工具函数 tool_fn = tool_map[tool_name] # 执行工具（这里是真正的函数调用） tool_result = tool_fn.invoke(tool_args) print(f"工具 {tool_name} 执行结果：{tool_result}") # 把工具结果封装成 ToolMessage，加入消息历史 # tool_call_id 要和 AIMessage 里对应的 id 匹配，模型靠这个对应结果 messages.append(ToolMessage( content=str(tool_result), tool_call_id=tool_id, # 关键：必须和 AIMessage.tool_calls 里的 id 一致 )) print() # -------- 第三步：把工具结果发回给模型，得到最终回答 -------- final_response = llm_with_tools.invoke(messages) print("最终回答：") print(final_response.content) # 完整的消息历史就是： # 1. HumanMessage（用户问题） # 2. AIMessage（模型的工具调用决策） # 3. ToolMessage（工具1的执行结果） # 4. ToolMessage（工具2的执行结果） # 5. AIMessage（基于工具结果的最终回答）
更复杂的场景下，模型可能需要多次调用工具（调用工具A → 根据结果再决定调用工具B → 综合结果回答）。这就是 ReAct（Reasoning + Acting）模式的核心：
from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from typing import List # 定义一组互相配合的工具 @tool def search_employee(name: str) -> str: """ 根据员工姓名查找员工 ID 和基本信息。 当需要获取员工 ID 以进行后续查询时，先用此工具。 """ employees = { "张伟": {"id": "EMP001", "department": "技术部", "position": "高级工程师"}, "李娜": {"id": "EMP002", "department": "产品部", "position": "产品经理"}, "王强": {"id": "EMP003", "department": "销售部", "position": "销售总监"}, } emp = employees.get(name) if emp: return f"找到员工：{name}，ID：{emp['id']}，部门：{emp['department']}，职位：{emp['position']}" return f"未找到员工：{name}" @tool def get_employee_salary(employee_id: str) -> str: """ 根据员工 ID 查询薪资信息。 需要先通过 search_employee 获取员工 ID 才能使用此工具。 """ salaries = { "EMP001": {"base": 25000, "bonus": 5000, "total": 30000}, "EMP002": {"base": 22000, "bonus": 4000, "total": 26000}, "EMP003": {"base": 20000, "bonus": 8000, "total": 28000}, } salary = salaries.get(employee_id) if salary: return f"员工 {employee_id} 薪资：底薪 {salary['base']} 元，奖金 {salary['bonus']} 元，税前总计 {salary['total']} 元" return f"未找到员工 {employee_id} 的薪资记录" @tool def get_department_budget(department: str) -> str: """查询部门的年度预算使用情况。""" budgets = { "技术部": {"total": 500000, "used": 320000, "remaining": 180000}, "产品部": {"total": 300000, "used": 150000, "remaining": 150000}, "销售部": {"total": 800000, "used": 600000, "remaining": 200000}, } budget = budgets.get(department) if budget: usage_rate = budget['used'] / budget['total'] * 100 return (f"{department}年度预算：{budget['total']} 元，" f"已使用 {budget['used']} 元（{usage_rate:.1f}%），" f"剩余 {budget['remaining']} 元") return f"未找到 {department} 的预算数据" # 工具列表和查找字典 tools = [search_employee, get_employee_salary, get_department_budget] tool_map = {t.name: t for t in tools} llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) def run_tool_loop(user_question: str, max_iterations: int = 5) -> str: """ 手动实现的工具调用循环（ReAct 模式） 这个函数展示了 AgentExecutor 内部的核心逻辑。 每次迭代： 1. 把当前消息列表发给模型 2. 如果模型决定调用工具 → 执行工具，把结果加入消息列表，继续循环 3. 如果模型给出最终回答（没有 tool_calls）→ 返回回答，退出循环 """ messages = [HumanMessage(content=user_question)] print(f"用户问：{user_question}") print("=" * 60) for iteration in range(max_iterations): print(f"\n[第 {iteration + 1} 轮推理]") # 发给模型 response = llm_with_tools.invoke(messages) messages.append(response) # 判断模型是否需要调用工具 if not response.tool_calls: # 没有 tool_calls，说明模型给出了最终回答 print("模型给出最终回答（不再需要工具）") return response.content # 有 tool_calls，执行所有工具 print(f"模型决定调用 {len(response.tool_calls)} 个工具：") for call in response.tool_calls: print(f" → {call['name']}({call['args']})") # 执行工具 tool_fn = tool_map.get(call["name"]) if tool_fn is None: tool_result = f"错误：工具 '{call['name']}' 不存在" else: try: tool_result = tool_fn.invoke(call["args"]) except Exception as e: tool_result = f"工具执行失败：{e}" print(f" 结果：{tool_result}") # 把工具结果加入消息历史 messages.append(ToolMessage( content=str(tool_result), tool_call_id=call["id"], )) # 超过最大迭代次数，强制返回 return "抱歉，处理超时，无法给出完整回答。" # 测试：一个需要多轮工具调用的复杂问题 answer = run_tool_loop("查一下张伟的薪资，以及他所在部门的预算情况") print("\n" + "=" * 60) print("最终回答：") print(answer) # 这个问题的工具调用链： # 第 1 轮：调用 search_employee("张伟") → 得到 ID=EMP001，部门=技术部 # 第 2 轮：同时调用 get_employee_salary("EMP001") 和 get_department_budget("技术部") # 第 3 轮：模型综合结果，给出最终自然语言回答
这段代码展示了整个 ReAct 循环的精髓：每次迭代，模型根据当前拥有的信息（包含之前工具的执行结果）来决定下一步行动，直到它认为信息足够充分才给出最终回答。
工具在执行过程中会遇到各种错误：网络超时、参数无效、权限不足……生产环境必须有完善的错误处理机制。
from langchain_core.tools import tool from typing import Union import requests @tool def fetch_url_content(url: str) -> str: """ 获取指定 URL 的网页内容摘要。 输入完整的 URL（需要包含 http:// 或 https://），返回页面的主要文本内容。 """ # 基本的 URL 验证 if not url.startswith(("http://", "https://")): # 返回明确的错误信息，而不是抛出异常 # 这样模型可以理解出了什么问题，并可能尝试修正参数 return f"错误：URL 格式不正确，必须以 http:// 或 https:// 开头。收到的 URL：{url}" try: response = requests.get(url, timeout=10) response.raise_for_status() # 如果状态码不是 200，抛出异常 # 简化处理：只返回前 500 个字符 return f"页面内容（前 500 字）：{response.text[:500]}" except requests.exceptions.Timeout: return f"错误：请求超时，网站 {url} 响应太慢（超过 10 秒）" except requests.exceptions.HTTPError as e: return f"错误：HTTP 请求失败，状态码 {e.response.status_code}，URL：{url}" except requests.exceptions.ConnectionError: return f"错误：无法连接到 {url}，请检查网址是否正确或网络连接" except Exception as e: return f"错误：获取内容时发生未预期的错误：{str(e)}"
关键原则：工具在出错时，应该返回描述性的错误信息字符串，而不是抛出异常。这样模型能看到错误内容，并有机会用不同的参数重试，或者告知用户具体的问题所在。
LangChain 提供了 handle_tool_error 参数，可以在 AgentExecutor 层面统一处理工具异常：
handle_tool_error
from langchain_core.tools import tool, ToolException @tool def divide_numbers(a: float, b: float) -> float: """ 计算 a 除以 b 的结果。 如果 b 为 0，会抛出错误。 """ if b == 0: # 使用 ToolException 明确标识这是工具错误 raise ToolException("除数不能为零！请提供一个非零的除数。") return a / b @tool def access_secret_file(filename: str) -> str: """读取文件内容。""" if "secret" in filename.lower(): raise ToolException(f"权限拒绝：无权访问文件 '{filename}'，该文件属于受保护资源。") return f"文件 {filename} 的内容：（模拟内容）" # 在 AgentExecutor 中配置错误处理 from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") tools = [divide_numbers, access_secret_file] prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有帮助的助手。"), ("human", "{input}"), MessagesPlaceholder("agent_scratchpad"), ]) agent = create_tool_calling_agent(llm, tools, prompt) # handle_tool_error=True：工具出错时，把错误信息发回给模型，让模型决定下一步 agent_executor = AgentExecutor( agent=agent, tools=tools, handle_tool_error=True, # 工具异常不会中断执行，而是作为工具结果返回 verbose=True, ) result = agent_executor.invoke({"input": "帮我计算 10 除以 0"}) print(result["output"]) # 模型会收到"除数不能为零"的错误信息，然后告知用户这个问题
对于可能长时间运行的工具，需要设置超时：
from langchain_core.tools import tool import signal from contextlib import contextmanager @contextmanager def timeout(seconds: int): """超时上下文管理器""" def handler(signum, frame): raise TimeoutError(f"工具执行超时（超过 {seconds} 秒）") old_handler = signal.signal(signal.SIGALRM, handler) signal.alarm(seconds) try: yield finally: signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) @tool def slow_analysis(data: str) -> str: """ 对大型数据集进行复杂分析（可能需要较长时间）。 超时时间：30 秒。 """ try: with timeout(30): # 模拟耗时操作 import time time.sleep(2) # 正常情况 return f"分析完成：{data[:50]} 的分析结果（模拟）" except TimeoutError as e: return f"分析超时：{e}。建议减少数据量或拆分任务。"
当你的应用需要大量工具时，工具集的设计质量直接影响模型的调用准确率。
from langchain_core.tools import tool # 原则一：工具名要动词开头，清晰描述动作 # ❌ 不好 @tool def weather(city: str) -> str: ... # 名词，不清晰 @tool def city_weather_information(city: str) -> str: ... # 太长 # ✅ 好 @tool def get_weather(city: str) -> str: ... # get + 名词 @tool def search_products(keyword: str) -> str: ... # search + 名词 @tool def create_calendar_event(title: str, date: str) -> str: ... # create + 名词 # 原则二：相似功能的工具要在描述中明确区分 @tool def search_web(query: str) -> str: """ 在互联网上搜索实时信息和最新新闻。 适用于：最新事件、实时数据、近期新闻、不确定的事实信息。 【注意】如果是查询内部公司数据，请使用 search_internal_docs 工具。 """ ... @tool def search_internal_docs(query: str) -> str: """ 在公司内部知识库中搜索文档和资料。 适用于：公司政策、产品文档、内部流程、历史项目资料。 【注意】如果是查询互联网上的公开信息，请使用 search_web 工具。 """ ... # 原则三：工具数量不宜过多（一般不超过 15 个） # 工具太多会让模型"选择困难"，调用准确率下降 # 解决方案：把相关工具分组，通过子 Agent 或路由来管理
当工具数量超过 10 个时，可以按职责分组，根据对话上下文动态加载相关工具：
from langchain_core.tools import tool from typing import List # 按业务领域分组 WEATHER_TOOLS = [get_weather] # 天气相关 CALENDAR_TOOLS = [] # 日历相关（略） EMAIL_TOOLS = [] # 邮件相关（略） DATABASE_TOOLS = [] # 数据库相关（略） # 工具注册表 TOOL_REGISTRY = { "weather": WEATHER_TOOLS, "calendar": CALENDAR_TOOLS, "email": EMAIL_TOOLS, "database": DATABASE_TOOLS, } def get_tools_for_context(user_message: str) -> List: """ 根据用户消息内容，动态选择要加载的工具组。 减少每次请求中的工具数量，提高模型的调用准确率。 """ tools = [] message_lower = user_message.lower() if any(kw in message_lower for kw in ["天气", "气温", "下雨", "温度"]): tools.extend(TOOL_REGISTRY["weather"]) if any(kw in message_lower for kw in ["日程", "会议", "预约", "提醒"]): tools.extend(TOOL_REGISTRY["calendar"]) if any(kw in message_lower for kw in ["邮件", "发送", "收件箱"]): tools.extend(TOOL_REGISTRY["email"]) if any(kw in message_lower for kw in ["查询", "数据", "记录", "订单"]): tools.extend(TOOL_REGISTRY["database"]) # 如果没有匹配，返回基础工具集 if not tools: tools = WEATHER_TOOLS # 默认工具 return tools
把本篇所有知识点整合，构建一个真实可用的多工具客服系统：
import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.tools import tool from langchain_core.messages import HumanMessage, AIMessage, ToolMessage, SystemMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory load_dotenv() # ============================================================ # 工具定义 # ============================================================ @tool def check_order_status(order_id: str) -> str: """ 查询订单状态和物流信息。 当用户询问订单进展、物流到哪里了、预计什么时候到达时使用此工具。 输入订单号（格式：ORD + 6位数字，例如 ORD123456）。 """ # 模拟订单数据库 orders = { "ORD123456": { "status": "运输中", "product": "蓝牙耳机 Pro", "order_date": "2025-05-20", "estimated_arrival": "2025-05-25", "logistics": "顺丰速运 SF1234567890", "current_location": "上海中转站", }, "ORD654321": { "status": "已签收", "product": "机械键盘", "order_date": "2025-05-15", "estimated_arrival": "2025-05-20", "logistics": "京东物流 JD9876543210", "current_location": "已送达", }, } order = orders.get(order_id.upper()) if not order: return f"未找到订单号 {order_id}，请确认订单号是否正确（格式：ORD + 6位数字）" return ( f"订单 {order_id} 详情：\n" f"商品：{order['product']}\n" f"状态：{order['status']}\n" f"下单日期：{order['order_date']}\n" f"预计到达：{order['estimated_arrival']}\n" f"物流单号：{order['logistics']}\n" f"当前位置：{order['current_location']}" ) @tool def search_products( keyword: str, max_price: float = None, category: str = None, ) -> str: """ 搜索商品信息，包含价格、库存和评分。 当用户询问有没有某种商品、某商品多少钱、推荐什么产品时使用。 可以按关键词、价格上限、商品分类进行筛选。 """ products = [ {"name": "蓝牙耳机 Pro", "price": 599, "category": "耳机", "stock": 50, "rating": 4.8}, {"name": "蓝牙耳机 Lite", "price": 299, "category": "耳机", "stock": 100, "rating": 4.5}, {"name": "机械键盘 RGB", "price": 399, "category": "键盘", "stock": 30, "rating": 4.7}, {"name": "无线鼠标", "price": 199, "category": "鼠标", "stock": 0, "rating": 4.3}, {"name": "显示器 27寸", "price": 1299, "category": "显示器", "stock": 15, "rating": 4.9}, ] # 过滤 results = [p for p in products if keyword.lower() in p["name"].lower()] if max_price: results = [p for p in results if p["price"] <= max_price] if category: results = [p for p in results if p["category"] == category] if not results: return f"未找到符合条件的商品（关键词：{keyword}）" result_text = f"找到 {len(results)} 款商品：\n" for p in results: stock_text = f"库存 {p['stock']} 件" if p['stock'] > 0 else "暂时缺货" result_text += ( f"- {p['name']}：¥{p['price']}，{stock_text}，" f"评分 {p['rating']}/5.0\n" ) return result_text.strip() @tool def apply_for_refund(order_id: str, reason: str) -> str: """ 申请退款或退货。 当用户表达不满、要求退款、商品有问题需要退换货时使用此工具。 需要提供订单号和退款原因。退款申请提交后通常 3-5 个工作日处理。 """ valid_reasons = ["质量问题", "与描述不符", "不想要了", "收到时已损坏", "发错货"] # 验证原因（实际系统中会更复杂） is_valid_reason = any(r in reason for r in valid_reasons) if not is_valid_reason: return ( f"退款原因不够清晰，请选择以下原因之一：\n" + "\n".join(f"- {r}" for r in valid_reasons) ) return ( f"退款申请已提交！\n" f"订单号：{order_id}\n" f"退款原因：{reason}\n" f"申请编号：RF{order_id[-6:]}，请保存此编号\n" f"预计 3-5 个工作日内处理，退款将原路返回" ) @tool def get_promotion_info(product_category: str = None) -> str: """ 获取当前促销活动和优惠券信息。 当用户询问有没有折扣、优惠活动、优惠码时使用此工具。 可以按商品分类查询对应的促销，不指定分类则返回全场促销。 """ promotions = { "全场": "618 活动进行中！满 500 减 80，满 1000 减 200，优惠码：618SAVE", "耳机": "耳机专区额外 9 折，叠加全场优惠，优惠码：HEADPHONE10", "键盘": "键盘鼠标套装购买享 8.5 折，优惠码：KEYBOARD15", } if product_category and product_category in promotions: return f"{product_category}类促销：{promotions[product_category]}\n全场通用：{promotions['全场']}" all_promos = "\n".join([f"- {k}：{v}" for k, v in promotions.items()]) return f"当前促销活动：\n{all_promos}" # ============================================================ # 构建带记忆的多工具客服链 # ============================================================ tools = [check_order_status, search_products, apply_for_refund, get_promotion_info] tool_map = {t.name: t for t in tools} llm = ChatOpenAI(model="gpt-4o", temperature=0.3) llm_with_tools = llm.bind_tools(tools) SYSTEM_PROMPT = """你是一位专业、友善的电商客服助手。 你的职责： - 帮助用户查询订单状态和物流 - 协助查找和推荐商品 - 处理退款申请 - 提供促销和优惠信息 服务原则： - 称呼用户时用"您"，保持礼貌 - 如果需要信息（如订单号）才能完成任务，主动询问 - 退款场景下，先表达理解，再提供解决方案 - 回答简洁明了，避免不必要的废话 """ session_store = {} def get_session_history(session_id: str) -> InMemoryChatMessageHistory: if session_id not in session_store: session_store[session_id] = InMemoryChatMessageHistory() return session_store[session_id] def chat_with_tools(user_input: str, session_id: str, max_tool_rounds: int = 5) -> str: """ 带工具调用能力的多轮对话函数。 支持历史记忆，每次对话自动维护上下文。 """ history = get_session_history(session_id) # 构建完整的消息列表：系统提示 + 历史 + 当前输入 messages = [SystemMessage(content=SYSTEM_PROMPT)] messages.extend(history.messages) messages.append(HumanMessage(content=user_input)) # 工具调用循环 for _ in range(max_tool_rounds): response = llm_with_tools.invoke(messages) messages.append(response) # 如果没有工具调用，直接返回 if not response.tool_calls: # 更新历史（只保存 human 和 final ai 消息） history.add_user_message(user_input) history.add_ai_message(response.content) return response.content # 执行所有工具 for call in response.tool_calls: tool_fn = tool_map.get(call["name"]) try: result = tool_fn.invoke(call["args"]) if tool_fn else f"工具 {call['name']} 不存在" except Exception as e: result = f"工具执行出错：{e}" messages.append(ToolMessage( content=str(result), tool_call_id=call["id"], )) return "抱歉，处理您的请求时遇到了困难，请稍后重试或联系人工客服。" # ============================================================ # 演示对话 # ============================================================ if __name__ == "__main__": session = "demo_user_001" conversations = [ "你好！我想查一下我的订单，单号是 ORD123456", "好的，那有没有什么蓝牙耳机推荐？预算在 400 以内", "有优惠活动吗？", "那个蓝牙耳机 Lite 收到货质量有问题，我要退款", ] print("=" * 60) print("客服系统演示开始") print("=" * 60) for user_msg in conversations: print(f"\n用户：{user_msg}") reply = chat_with_tools(user_msg, session) print(f"客服：{reply}") print("-" * 40)
你现在应该能做到： - 用 @tool 和 StructuredTool 定义各类工具，写出让模型准确调用的工具描述 - 用 bind_tools() 把工具注册给模型，理解 Function Call 的底层机制 - 手动实现完整的工具调用循环（ReAct 模式），不依赖 AgentExecutor - 为工具设计健壮的错误处理，避免工具失败导致整个对话崩溃 - 构建生产可用的多工具对话系统
《Agent 开发——让模型自主决策调用工具》
掌握了手动工具调用之后，第八篇将介绍 LangChain 提供的自动化方案——Agent：
create_tool_calling_agent
create_react_agent
AgentExecutor
代码仓库：本系列所有可运行代码示例统一维护在 GitHub，每篇对应独立目录，可直接克隆运行。 系列导航：[第一篇] → ... → [第六篇] → 第七篇（当前） → 第八篇 → ...
代码仓库：本系列所有可运行代码示例统一维护在 GitHub，每篇对应独立目录，可直接克隆运行。
系列导航：[第一篇] → ... → [第六篇] → 第七篇（当前） → 第八篇 → ...
还没有评论，来抢沙发吧！
博客管理员
40 篇文章
admin
还没有评论，来抢沙发吧！