---

前言

提示：前面7章内容已介绍完成的应用搭建步骤，本文主要介绍大模型能力扩展（通过工具实现）和应用部署（LangServe）：

在前面的对话基础上，设想一个场景，如果用户问大模型“XXX城市今天的天气怎么样？”，通过前面的内容我们知道，通用大模型是没有这种实时数据的，只能通过其他方式查询天气数据，然后增强提示词，交给大模型才行。

传统解决办法：通过调用获取天气的接口，拿到数据后增强提示词。确实解决了这个问题。

但是，用户的提问总是多样的，这个时候，又有用户问“公司在XXX城市的考勤情况是怎么样的？”，大模型又无法解答了，这个时候又要去新增调用考勤情况的方法才行。

根据上述场景，大家就能发现这种方式太机械了，不符合实际的应用场景，有无办法提前将这些与外部交互的方法写好，然后当大模型识别到用户的提问后，就调用对应的方法获取数据仅回答？

有的，有的，兄弟有的！！！

这就是本文要介绍的工具的作用，我们将提前写好的这些方法工具化，并告知大模型，这个工具是用来做什么的，然后大模型就会在需要这些能力的时候自动调用这些工具

---

一、工具是什么？

大模型的能力从前面的学习，大家肯定知道，就是文本的识别和内容生成。就发现大模型像一个智慧大脑，可以去思考和回答。但是“它”没有四肢。举例：大模型想到这个地方需要去查天气数据（想象成一本书），但是它发现自己没法翻开这本书。

这个时候造物主（即“程序员”），给大模型安装上了多功能手脚（即工具），这个时候，大模型就能够通过这些手脚去翻开天气这本书，获取到里面的数据。

所以大家可以简单理解工具就是：赋予大模型“执行”能力的东西，让大模型完成从只能思考到“思考+执行”的升级

工具能干嘛？你能用程序实现的一切事情。（查数据、统计数据、生成视频、生成图片等等。题外话：从这就可以看出来，大模型生成视频这些内容本质上还是用的生成视频的工具，即外部系统的能力，而非它本身具备这个能力）

二、工具的创建及使用

整个流程：

演示代码：

import os
from typing import Type
from pydantic import BaseModel, Field
#所有自定义的tool都需要继承BaseTool，并实现_run方法
from langchain.tools import BaseTool, tool
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage
from langchain_core.prompts import PromptTemplate

# 加载环境变量
load_dotenv()
MODULE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
MODULE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODULE_NAME = "qwen-plus"

# 初始化通义千问大模型
llm = ChatOpenAI(
    model=MODULE_NAME,
    api_key=MODULE_API_KEY,
    base_url=MODULE_BASE_URL,
    temperature=0.1,
    max_tokens=1024
)

# 方式1：使用@tool装饰器定义天气查询工具（简化写法）
@tool
def custom_weather_tool(city: str) -> str:
    """
    自定义天气查询工具，输入城市名称，返回该城市的天气信息。
    参数：city - 城市名称（字符串，如北京、上海）
    """
    # 模拟调用天气API
    weather_data = {
        "北京": "晴，25℃",
        "上海": "多云，23℃",
        "广州": "雷阵雨，28℃"
    }
    return f"{city}的天气：{weather_data.get(city, '未查询到该城市天气')}"

# 方式2：继承BaseTool类定义计算器工具（修复属性赋值方式）

class CalculatorInput(BaseModel):
    """计算器工具的输入参数模型"""
    expression: str = Field(description="需要计算的数学表达式，如'10+20*3'或'9/3-2'")

class CustomCalculatorTool(BaseTool):
    # 直接初始化实例属性（无需ClassVar，兼容所有版本）
    name = "custom_calculator"
    description = "用于执行数学计算，输入合法的数学表达式，返回计算结果"
    #继承BaseTool的属性，用于固定该工具类的输入参数的格式，并按格式对输入参数进行校验，校验规则为description的描述
    args_schema: Type[BaseModel] = CalculatorInput

    def _run(self, expression: str) -> str:
        """同步执行逻辑：计算数学表达式"""
        try:
            result = eval(expression)
            return f"计算结果：{expression} = {result}"
        except Exception as e:
            return f"计算失败：{str(e)}，请输入合法的数学表达式"

    def _arun(self, expression: str):
        """异步执行逻辑（本工具不支持异步，抛出异常）"""
        raise NotImplementedError("该工具不支持异步调用")

if __name__ == "__main__":
    # 注册工具列表（实例化类工具）
    tools = [custom_weather_tool, CustomCalculatorTool()]

    # 定义Agent提示词（必须包含agent_scratchpad占位符）
    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content="你是一个智能助手，会根据需求调用合适的工具完成任务。"),
        ("user", "{input}"),  # 用户输入
        # 必须用MessagesPlaceholder存储调用工具的思考过程，否则工具会调用失败
        MessagesPlaceholder(variable_name="agent_scratchpad"),  
    # 创建OpenAI Tools Agent
    agent = create_openai_tools_agent(llm, tools, prompt)

    # 创建Agent执行器。将大模型、工具结合起来使用
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose= false,  # 改成true开启详细日志，可以查看工具调用过程
        handle_parsing_errors=lambda e: f"解析错误：{str(e)}，请重新描述你的需求",
        max_iterations=3,
        early_stopping_method="force"
    )

    # 需要计算时就会调用计算的工具类完成业务
    print("===== 计算10+20*3 =====")
    result1 = agent_executor.invoke({"input": "请计算10+20*3"})
    print(result1["output"])

    print("===== 计算10+20和3 =====")
    result1 = agent_executor.invoke({"input": "请计算你好啊"})
    print(result1["output"])
    
    #需要查询天气时，就会调用天气查询的工具
    print("\n===== 查询北京天气 =====")
    result2 = agent_executor.invoke({"input": "请查询北京的天气"})
    print(result2["output"])

    print("\n===== 查询深圳天气 =====")
    result3 = agent_executor.invoke({"input": "请查询深圳的天气"})
    print(result3["output"])

执行结果：

小结：

工具的使用步骤：
1、创建工具。有2中方法：①装饰器@tool ②继承BaseTool并实现-run方法
2、注册工具。即实体化工具类
3、创建智能体执行器AgentExecutor。它负责协调 Agent 的思考、工具调用、结果处理和迭代过程，是连接用户输入、Agent 决策、工具执行的核心枢纽

AgentExecutor的整个流程为（感兴趣的可以自己去学习源码，并尝试自己实现这个步骤）：
1、接受入参，如：用户提问，然后调用大模型，生成决策。
2、根据决策，从工具列表里查找用哪个工具执行
3、将工具执行的结果返回

---

三、实例：根据用户提问调用数据库工具类，实现数据库数据获取

import os
import pymysql
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import Type
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import BaseTool
from langchain_core.messages import SystemMessage

# 加载环境变量
load_dotenv()

MODULE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
MODULE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODULE_NAME = "qwen-plus"

# 初始化大模型
llm = ChatOpenAI(
    model=MODULE_NAME,
    api_key=MODULE_API_KEY,
    base_url=MODULE_BASE_URL,
    temperature=0.1,
    max_tokens=1024
)


# SQL生成工具
class SqlGenerateInput(BaseModel):
    query: str = Field(description="用于生成SQL语句的自然语言查询文本")

class SqlGenerateTool(BaseTool):
    name = "sql_generate"
    description = "用于生成SQL语句，输入自然语言查询文本，返回对应的MySQL SELECT语句"
    args_schema: Type[BaseModel] = SqlGenerateInput

    def _run(self, query: str) -> str:
        # 优化数据库表结构描述，提升SQL生成准确性
        prompt = PromptTemplate(
            input_variables=["nl_query"],
            template="""你是专业的MySQL SQL生成助手，需根据自然语言查询和指定表结构生成准确的SELECT语句。
            要求：
            1. 仅返回SQL语句，无任何解释性文字
            2. 严格匹配表名和字段名，表名：products，数据库名：lctest
            3. 仅生成SELECT查询语句，禁止写操作
            4. 条件不明确时生成合理默认查询
            
            数据库表结构：
            lctest数据库的products表包含以下字段：
            - id: int，主键，非空，产品唯一ID
            - product_name: varchar(100)，非空，产品名称
            - price: decimal(10,2)，非空，产品价格（保留2位小数）
            - stock: int，非空，产品库存
            
            用户自然语言查询：{nl_query}
            生成的SQL语句："""
        )
        chain = prompt | llm | StrOutputParser()
        sql = chain.invoke({"nl_query": query})
        # 清理生成的SQL中的多余字符（如换行、空格）
        return sql.strip()


# SQL执行工具
class SqlExecuteInput(BaseModel):
    sql: str = Field(description="需要执行的MySQL SELECT语句")

class SqlExecuteTool(BaseTool):
    name = "sql_execute"
    description = "用于执行MySQL SELECT语句，输入SQL语句，返回执行结果"
    args_schema: Type[BaseModel] = SqlExecuteInput

    def _run(self, sql: str) -> str:
        conn = None
        cursor = None
        try:
            # 连接数据库（从环境变量获取配置）
            conn = pymysql.connect(
                host=os.getenv("MYSQL_HOST", "localhost"),
                port=int(os.getenv("MYSQL_PORT", "3306")),
                user=os.getenv("MYSQL_USER", "root"),
                password=os.getenv("MYSQL_PASSWORD", "123456"),
                database=os.getenv("MYSQL_DATABASE", "lctest"),
                charset="utf8mb4"
            )
            cursor = conn.cursor()
            cursor.execute(sql)
            # 获取字段名和结果集，让返回更易读
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            result = cursor.fetchall()
            # 格式化结果
            if not result:
                return "查询结果为空"
            return f"字段：{columns}\n查询结果：{result}"
        except Exception as e:
            return f"SQL执行失败：{str(e)}"
        finally:
            # 安全关闭游标和连接（避免未初始化的情况）
            if cursor:
                cursor.close()
            if conn:
                conn.close()


# 注册工具（可根据需求选择是否添加执行工具）
tools = [SqlGenerateTool(), SqlExecuteTool()]

# 定义Agent提示词（必须包含agent_scratchpad占位符）
prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content="你是智能SQL助手，会根据用户需求调用sql_generate工具生成SQL语句，如需执行可调用sql_execute工具。"),
    ("user", "{nl_query}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),  # 工具调用中间步骤占位符
])

# 创建Agent
agent = create_openai_tools_agent(llm, tools, prompt)

# 创建Agent执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=False,  # 开启详细日志，便于调试工具调用过程
    handle_parsing_errors="返回解析错误，请重新描述查询需求",
    max_iterations=5,
    early_stopping_method="force"
)

# 测试：调用Agent（注意传入的参数名是input，不是query）
if __name__ == "__main__":
    # 测试：生成库存大于1000的产品查询SQL
    result1 = agent_executor.invoke({"nl_query": "查询库存大于1000的产品信息"})
    print("===== 测试1结果 =====")
    print(result1["output"])

运行结果：

数据库数据：

四、部署服务，让智能体能够服务

import os
import pymysql
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import Type, Dict, Any
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain.tools import BaseTool
from langchain_core.messages import SystemMessage
from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.runnables import RunnableConfig
from fastapi import FastAPI
import uvicorn

# 加载环境变量
load_dotenv()

# ====================== 1. 初始化大模型 ======================
MODULE_API_KEY = os.getenv("DASHSCOPE_API_KEY")
MODULE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
MODULE_NAME = "qwen-plus"

llm = ChatOpenAI(
    model=MODULE_NAME,
    api_key=MODULE_API_KEY,
    base_url=MODULE_BASE_URL,
    temperature=0.1,
    max_tokens=1024
)


# ====================== 2. 定义SQL工具 ======================
# SQL生成工具
class SqlGenerateInput(BaseModel):
    query: str = Field(description="用于生成SQL语句的自然语言查询文本")


class SqlGenerateTool(BaseTool):
    name = "sql_generate"
    description = "用于生成SQL语句，输入自然语言查询文本，返回对应的MySQL SELECT语句"
    args_schema: Type[BaseModel] = SqlGenerateInput

    def _run(self, query: str) -> str:
        prompt = PromptTemplate(
            input_variables=["nl_query"],
            template="""你是专业的MySQL SQL生成助手，需根据自然语言查询和指定表结构生成准确的SELECT语句。
            要求：
            1. 仅返回SQL语句，无任何解释性文字
            2. 严格匹配表名和字段名，表名：products，数据库名：lctest
            3. 仅生成SELECT查询语句，禁止写操作
            4. 条件不明确时生成合理默认查询

            数据库表结构：
            lctest数据库的products表包含以下字段：
            - id: int，主键，非空，产品唯一ID
            - product_name: varchar(100)，非空，产品名称
            - price: decimal(10,2)，非空，产品价格（保留2位小数）
            - stock: int，非空，产品库存

            用户自然语言查询：{nl_query}
            生成的SQL语句："""
        )
        chain = prompt | llm | StrOutputParser()
        sql = chain.invoke({"nl_query": query})
        return sql.strip()


# SQL执行工具
class SqlExecuteInput(BaseModel):
    sql: str = Field(description="需要执行的MySQL SELECT语句")


class SqlExecuteTool(BaseTool):
    name = "sql_execute"
    description = "用于执行MySQL SELECT语句，输入SQL语句，返回执行结果"
    args_schema: Type[BaseModel] = SqlExecuteInput

    def _run(self, sql: str) -> str:
        conn = None
        cursor = None
        try:
            conn = pymysql.connect(
                host=os.getenv("MYSQL_HOST", "localhost"),
                port=int(os.getenv("MYSQL_PORT", "3306")),
                user=os.getenv("MYSQL_USER", "root"),
                password=os.getenv("MYSQL_PASSWORD", "123456"),
                database=os.getenv("MYSQL_DATABASE", "lctest"),
                charset="utf8mb4"
            )
            cursor = conn.cursor()
            cursor.execute(sql)
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            result = cursor.fetchall()
            if not result:
                return "查询结果为空"
            return f"字段：{columns}\n查询结果：{result}"
        except Exception as e:
            return f"SQL执行失败：{str(e)}"
        finally:
            if cursor:
                cursor.close()
            if conn:
                conn.close()


# ====================== 3. 创建Agent ======================
# 注册工具
tools = [SqlGenerateTool(), SqlExecuteTool()]

# 定义Agent提示词
prompt = ChatPromptTemplate.from_messages([
    SystemMessage(
        content="你是智能SQL助手，会根据用户需求调用sql_generate工具生成SQL语句，如需执行可调用sql_execute工具。"),
    ("user", "{nl_query}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 创建Agent和执行器
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=False,
    handle_parsing_errors="返回解析错误，请重新描述查询需求",
    max_iterations=5,
    early_stopping_method="force"
)


# ====================== 4. 部署LangChain Server ======================
# 初始化FastAPI应用
app = FastAPI(title="SQL智能助手服务", version="1.0")

# 将Agent封装为Runnable并添加路由
# 定义API接口
@app.post("/sql-agent", summary="自然语言转SQL并执行")
async def sql_agent(nl_query: str):
    try:
        # 调用Agent
        result = agent_executor.invoke({"nl_query": nl_query})
        return {"code": 200, "message": "success", "data": result["output"]}
    except Exception as e:
        return {"code": 500, "message": f"服务异常：{str(e)}", "data": None}

# 启动服务的入口
if __name__ == "__main__":
    # 启动UVicorn服务
    uvicorn.run(
        "lc_server:app",  # 指向当前文件的app对象
        host="0.0.0.0",  # 允许外部访问
        port=8000,  # 服务端口
        reload=True  # 开发模式下自动重载
    )

使用fastapi自带的接口swagger界面进行接口测试

总结

1、通过本文需要掌握工具类的使用
2、需要掌握如何暴露应用的api

结尾：通过这8片的文章，基本已经了解了整个AIGC的流程及如何实现，后面大家可以自行找寻实战项目夯实下知识。本系列不进行实战项目讲解

后续内容将进入多模态相关内容学习