Agent 工具封装和调用

藏冰

255人浏览 · 2026-05-07 21:41:07

藏冰 · 2026-05-07 21:41:07 发布

LangChain Agent 工具封装和调用

Agent（智能体）是 LangChain 最核心的能力之一，它让大语言模型不再局限于文本生成，而是能够自主思考、调用工具、完成复杂任务。本文从工具调用的核心原理、工具封装规范、实战案例到企业级最佳实践，带你从零掌握 LangChain Agent 开发，附完整可运行代码。

一、Agent 与工具调用概述

1. 什么是 Agent

Agent 是一个能够自主感知环境、做出决策、执行动作的智能系统。在 LangChain 中，Agent = 大语言模型 + 工具集 + 推理框架：

大语言模型：作为“大脑”，负责思考和决策
工具集：作为“手脚”，负责执行具体操作（如API调用、计算、数据库查询）
推理框架：作为“神经中枢”，负责协调思考和执行的流程

2. 为什么需要工具调用

大语言模型本身存在天然局限，工具调用是解决这些问题的关键：

解决知识局限：大模型有知识截止日期，无法获取实时数据（如天气、股票、新闻）
消除幻觉问题：通过工具查询真实数据，避免模型编造虚假信息
扩展能力边界：让模型获得计算、API访问、文件操作、数据库查询等能力
模块化复用：将不同功能封装为独立工具，便于维护和扩展
完成复杂任务：通过多工具组合调用，解决需要多步骤的复杂问题（如“查询北京明天的天气，然后推荐适合的穿搭”）

二、工具封装基础：`@tool` 装饰器

LangChain 提供了 @tool 装饰器，能够快速将普通 Python 函数转换为 Agent 可调用的工具。这是最常用、最简洁的工具封装方式。

1. 核心语法

from langchain.tools import tool
import re

@tool
def function_name(input_str: str) -> str:
    """
    工具的详细描述（非常重要！大模型会根据这段描述决定是否调用该工具）
    参数说明：
        input_str: 工具需要的参数，必须明确说明格式和含义
    返回值：
        工具执行后的结果，大模型会基于这个结果继续思考
    """
    # 工具的具体实现逻辑
    return "执行结果"

关键注意事项：

函数注释必须清晰、准确，包含工具的作用、参数格式和返回值说明

参数类型和返回值类型必须明确标注（大模型会根据类型生成正确的参数）

工具名称要直观，让大模型一眼就能理解其功能

2. 工具元数据

封装后的工具会自动生成元数据，Agent 会根据这些元数据决定如何调用工具：

# 查看工具名称
print(multiply.name)
# 查看工具描述
print(multiply.description)
# 查看工具参数
print(multiply.args)
# 查看工具参数的 JSON Schema
print(multiply.args_schema.model_json_schema())

三、实战工具 1：乘法计算工具

我们先从一个简单的乘法计算工具开始，理解工具封装的完整流程。

1. 工具实现

from langchain.tools import tool
import re

@tool
def multiply(input_str: str) -> int:
    """
    计算两个整数的乘积
    参数：
        input_str: 包含两个整数的字符串，格式必须为 "a×b"，例如 "11×11"、"12×12"
    返回值：
        两个整数的乘积
    """
    # 使用正则提取字符串中的所有数字
    numbers = re.findall(r'\d+', input_str)
    if len(numbers) == 2:
        a, b = map(int, numbers)
        return a * b
    else:
        return "参数格式错误，请使用 'a×b' 的格式，例如 '11×11'"

2. 手动测试工具

在集成到 Agent 之前，先手动测试工具是否正常工作：

# 测试正常情况
print(multiply.invoke("11×11"))  # 输出：121
print(multiply.invoke("12×12"))  # 输出：144

# 测试错误格式
print(multiply.invoke("11*11"))  # 输出：参数格式错误，请使用 'a×b' 的格式，例如 '11×11'

四、实战工具 2：实时天气查询工具

接下来实现一个更实用的工具：调用第三方 API 查询实时天气。这个工具涉及城市代码匹配和API 调用，是企业级工具的典型代表。

1. 准备工作

（1）安装依赖

pip install requests pandas

（2）获取 API Key

访问 APISpace 天气 API，注册并获取你的 API Key
下载全国城市代码表 tianqi.csv（包含城市名称和对应的 areacode）

2. 实现城市代码匹配函数

天气 API 需要使用城市代码（areacode）查询，因此需要先实现一个函数，将城市名称转换为对应的代码：

import pandas as pd

# 加载城市代码表
city_df = pd.read_csv("tianqi.csv", sep='\t', encoding='UTF-8')

def get_city_code(city_name: str) -> str:
    """
    根据城市名称获取对应的城市代码（areacode）
    匹配优先级：区县 > 市 > 省
    参数：
        city_name: 城市名称，如 "北京"、"莲池区"、"保定"
    返回值：
        城市代码，默认返回保定市莲池区的代码 101090213
    """
    # 优先匹配区县
    match = city_df[city_df["district"] == city_name]
    if not match.empty:
        return match.iloc[0]["areacode/城市ID"]
    
    # 其次匹配市
    match = city_df[city_df["city"] == city_name]
    if not match.empty:
        return match.iloc[0]["areacode/城市ID"]
    
    # 最后模糊匹配省
    match = city_df[city_df["city"].str.contains(city_name, na=False)]
    if not match.empty:
        return match.iloc[0]["areacode/城市ID"]
    
    # 默认返回保定市莲池区
    return "101090213"

# 测试城市代码匹配
print(get_city_code("丰台"))    # 输出：101010900
print(get_city_code("竞秀区"))  # 输出：101090202
print(get_city_code("保定"))    # 输出：101090201

3. 封装天气查询工具

import requests
from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """
    查询指定城市的实时天气信息
    参数：
        city: 城市名称，支持省、市、区县，如 "北京"、"保定"、"莲池区"
    返回值：
        包含温度和天气状况的字符串，如 "保定市莲池区的温度是22.7°C，天气状况是晴"
    """
    # 获取城市代码
    city_code = get_city_code(city)
    
    # 构造 API 请求
    url = "https://eolink.o.apispace.com/456456/weather/v001/now"
    payload = {"areacode": city_code}
    headers = {
        "X-APISpace-Token": "你的APISpace Token",  # 替换为你的实际 Token
        "Content-Type": "application/json"
    }
    
    try:
        # 发送 GET 请求
        response = requests.get(url, params=payload, headers=headers, timeout=10)
        response.raise_for_status()  # 抛出 HTTP 错误
        data = response.json()
        
        # 解析返回结果
        if data.get("status") == 0:
            location = data["result"]["location"]["name"]
            temp = data["result"]["realtime"]["temp"]
            weather = data["result"]["realtime"]["text"]
            return f"{city}（{location}）的温度是{temp}°C，天气状况是{weather}"
        else:
            return f"查询天气失败：{data.get('msg', '未知错误')}"
    
    except Exception as e:
        return f"查询天气时发生错误：{str(e)}"

# 手动测试天气工具
print(get_weather.invoke("保定"))
# 输出示例：保定（莲池）的温度是22.7°C，天气状况是晴

五、ReAct Agent 集成：让大模型自主调用工具

ReAct（Reasoning + Acting）是目前最常用的 Agent 推理框架，它让大模型在思考过程中交替进行推理（Reasoning） 和行动（Acting），逐步解决问题。

1. 初始化大语言模型

我们使用智谱 AI 的 GLM-4 作为 Agent 的大脑：

import os
from langchain_community.chat_models import ChatZhipuAI

# 配置智谱 API Key（替换为你的实际 Key）
os.environ["ZHIPUAI_API_KEY"] = "你的智谱API Key"

# 初始化 GLM-4 模型
llm = ChatZhipuAI(
    model="glm-4",
    temperature=0.3,  # 工具调用场景建议使用较低的 temperature，提高准确性
    max_tokens=2048
)

2. 创建 ReAct Agent

from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

# 1. 定义工具列表（将我们封装的两个工具加入）
tools = [get_weather, multiply]

# 2. 获取 ReAct 提示词模板（从 LangChain Hub 拉取官方优化后的模板）
prompt = hub.pull("hwchase17/react")

# 3. 创建 ReAct Agent
agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=prompt
)

# 4. 创建 Agent 执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 开启详细日志，显示 Agent 的完整思考过程
    handle_parsing_errors=True  # 自动处理解析错误，避免程序崩溃
)

3. Agent 核心参数说明

参数	作用
`verbose=True`	打印 Agent 的完整思考过程（Thought、Action、Action Input、Observation），非常适合调试
`handle_parsing_errors=True`	当大模型输出的格式不符合要求时，自动处理错误，让 Agent 继续执行
`max_iterations=3`	限制 Agent 的最大思考步骤，避免无限循环

六、完整实战：Agent 多工具调用测试

现在我们测试 Agent 能否根据用户问题，自主选择正确的工具并执行。

1. 测试天气查询

# 查询竞秀区的天气
response = agent_executor.invoke({"input": "请告诉我保定市竞秀区今天的天气情况"})
print("\n最终回答：", response["output"])

执行过程日志（verbose=True）：

> Entering new AgentExecutor chain...
用户询问保定市竞秀区今天的天气情况。我需要使用 get_weather 工具来获取实时天气信息，参数是城市名称 "竞秀区"。

Action: get_weather
Action Input: 竞秀区
Observation: 竞秀区（竞秀）的温度是22.5°C，天气状况是晴
我已经获取到竞秀区的实时天气信息，可以直接回答用户的问题了。

Thought: I now know the final answer
Final Answer: 保定市竞秀区今天的天气是晴，温度22.5°C。

> Finished chain.
最终回答： 保定市竞秀区今天的天气是晴，温度22.5°C。

2. 测试乘法计算

# 计算 12×12
response = agent_executor.invoke({"input": "12乘以12等于多少？"})
print("\n最终回答：", response["output"])

执行过程日志：

> Entering new AgentExecutor chain...
用户想知道12乘以12等于多少。这是一个乘法计算问题，我需要使用 multiply 工具，参数格式为 "12×12"。

Action: multiply
Action Input: 12×12
Observation: 144
我已经得到了计算结果，可以直接回答用户。

Thought: I now know the final answer
Final Answer: 12乘以12等于144。

> Finished chain.
最终回答： 12乘以12等于144。

3. 批量测试多个问题

test_questions = [
    "今天北京的天气怎么样？",
    "15×15等于多少？",
    "上海今天多少度？",
    "25×4等于多少？"
]

for question in test_questions:
    print(f"\n=== 问题：{question} ===")
    response = agent_executor.invoke({"input": question})
    print(f"回答：{response['output']}")