05_工具调用和外部集成

发表于 2025-06-25 更新于 2025-10-28 分类于 rd ， python Changyan：本文字数： 1.1k 阅读时长 ≈ 4 分钟

让AI真正”动起来”！本节课学习如何让LLM调用外部工具和API，实现查询天气、计算数学、搜索数据库等实用功能。掌握Function Calling的原理和实践，构建能够与外部世界交互的智能Agent。

🎯 学习目标

学完本节课，你将能够：

理解工具调用（Function Calling）的原理
定义和注册工具函数
让LLM智能选择和调用工具
处理工具调用的结果
构建多工具协作的Agent

1. 什么是工具调用？

1.1 场景举例

用户问： “北京今天天气怎么样？”

传统LLM： “抱歉，我无法查询实时天气信息。”（因为没有联网能力）

带工具的LLM：

理解用户要查天气
调用 get_weather("北京") 工具
获取结果：{"city": "北京", "temp": 25, "weather": "晴天"}
回复：”北京今天天气晴朗，气温25度。”

1.2 工具调用的流程

1	用户输入 → LLM分析 → 决定调用工具 → 执行工具函数 → 返回结果 → LLM生成回复

1.3 核心概念

工具（Tool） = 一个Python函数 + 描述信息

def get_weather(city: str) -> dict:
    """
    查询城市天气
    
    Args:
        city: 城市名称
        
    Returns:
        包含天气信息的字典
    """
    # 实际调用天气API
    return {"city": city, "temp": 25, "weather": "晴天"}

关键点：

函数的 docstring 很重要，LLM会根据它决定是否调用
参数要有类型注解
返回值要明确

2. LangChain的工具系统

2.1 定义工具

方式1：使用@tool装饰器

from langchain.tools import tool

@tool
def calculator(expression: str) -> float:
    """
    计算数学表达式的值
    
    Args:
        expression: 数学表达式，如"2+3*4"
    
    Returns:
        计算结果
    """
    try:
        return eval(expression)
    except Exception as e:
        return f"计算错误: {e}"

方式2：使用Tool类

from langchain.tools import Tool

def search_database(query: str) -> str:
    """搜索数据库"""
    # 实际搜索逻辑
    return f"查询结果：{query}"

search_tool = Tool(
    name="database_search",
    func=search_database,
    description="搜索数据库，返回相关记录"
)

2.2 实战：多功能助手

创建 demo_16/lesson_05/01_multi_tool_agent.py：

"""
多功能助手：集成计算器、天气查询等工具
学习目标：理解工具定义和调用流程
"""
import os
from typing import TypedDict, List
from typing_extensions import NotRequired
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
from langchain_community.chat_models import QianfanChatEndpoint
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from dotenv import load_dotenv

load_dotenv()


# 定义工具1：计算器
@tool
def calculator(expression: str) -> str:
    """
    计算数学表达式的值。
    
    支持基本的加减乘除运算。
    例如：2+3*4, 100/5, (2+3)*4
    
    Args:
        expression: 要计算的数学表达式
        
    Returns:
        计算结果的字符串
    """
    try:
        result = eval(expression)
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"


# 定义工具2：天气查询（模拟）
@tool
def get_weather(city: str) -> str:
    """
    查询指定城市的天气情况。
    
    Args:
        city: 城市名称，如"北京"、"上海"
        
    Returns:
        天气信息的描述
    """
    # 模拟天气数据
    weather_data = {
        "北京": "晴天，25度，空气质量良好",
        "上海": "多云，28度，湿度较高",
        "深圳": "雷阵雨，30度，注意防雨"
    }
    
    weather = weather_data.get(city, f"{city}的天气信息暂时无法获取")
    return f"{city}天气：{weather}"


# 定义工具3：数据库查询（模拟）
@tool
def query_database(table: str, condition: str) -> str:
    """
    查询数据库表。
    
    Args:
        table: 表名，如"users"、"orders"
        condition: 查询条件，如"age>18"
        
    Returns:
        查询结果
    """
    # 模拟数据库查询
    return f"从{table}表查询{condition}的记录：找到3条记录"


# 创建工具列表
tools = [calculator, get_weather, query_database]


# 定义状态
class State(TypedDict):
    messages: List
    tool_calls: NotRequired[List]


# 节点1：LLM决策
def llm_node(state: State):
    """LLM分析用户输入，决定是否调用工具"""
    llm = QianfanChatEndpoint(
        model="ERNIE-Speed-128K",
        qianfan_ak=os.getenv("QIANFAN_AK"),
        qianfan_sk=os.getenv("QIANFAN_SK"),
    )
    
    # 绑定工具到LLM
    llm_with_tools = llm.bind_tools(tools)
    
    # 调用LLM
    response = llm_with_tools.invoke(state["messages"])
    
    return {"messages": [response]}


# 节点2：执行工具
tool_node = ToolNode(tools)


# 路由：决定是调用工具还是结束
def should_continue(state: State) -> str:
    """判断是否需要调用工具"""
    last_message = state["messages"][-1]
    
    # 检查LLM是否要调用工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    else:
        return "end"


# 构建图
def create_graph():
    """创建工具调用图"""
    graph = StateGraph(State)
    
    # 添加节点
    graph.add_node("llm", llm_node)
    graph.add_node("tools", tool_node)
    
    # 设置流程
    graph.set_entry_point("llm")
    
    # LLM → 判断是否调用工具
    graph.add_conditional_edges(
        "llm",
        should_continue,
        {
            "tools": "tools",
            "end": END
        }
    )
    
    # 工具执行完 → 回到LLM
    graph.add_edge("tools", "llm")
    
    return graph.compile()


def main():
    """测试多工具助手"""
    app = create_graph()
    
    test_queries = [
        "计算 123 * 456 等于多少？",
        "查询一下北京的天气",
        "从users表查询age>25的记录",
    ]
    
    print("="*60)
    print("多功能AI助手")
    print("="*60)
    
    for query in test_queries:
        print(f"\n{'='*60}")
        print(f"用户: {query}")
        print(f"{'='*60}")
        
        # 执行
        result = app.invoke({
            "messages": [HumanMessage(content=query)]
        })
        
        # 打印最终回复
        final_message = result["messages"][-1]
        print(f"助手: {final_message.content}")


if __name__ == "__main__":
    main()