目录

1. 环境准备（终端）

2. MySQL建表与填入测试数据

3. 编写带参数的成绩查询 Tool

4.构建 Agent 自动调用

5.连接数据库并测试连接

6.测试运行

1. 环境准备（终端）

pip install langchain langchain-openai pymysql

2. MySQL建表与填入测试数据

use tool_demo;

-- 学生表
CREATE TABLE student (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(20) NOT NULL,
    gender VARCHAR(4),
    student_id VARCHAR(20) UNIQUE NOT NULL
);

-- 成绩表
CREATE TABLE score (
    id INT PRIMARY KEY AUTO_INCREMENT,
    subject VARCHAR(20) NOT NULL,
    exam_batch VARCHAR(20) NOT NULL,
    score INT NOT NULL,
    student_id INT NOT NULL,
    FOREIGN KEY (student_id) REFERENCES student(id)
);

-- 插入测试数据
INSERT INTO student(name, gender, student_id) 
VALUES ('张三','男','2026001'), ('李四','女','2026002');

INSERT INTO score(subject, exam_batch, score, student_id)
VALUES 
('数学','2026-01',90,1),
('语文','2026-01',85,1),
('数学','2026-01',88,2),
('英语','2026-01',92,2),
('数学','2026-02',96,1),
('语文','2026-02',87,1),
('数学','2026-02',89,2),
('英语','2026-02',96,2);

3. 编写带参数的成绩查询 Tool

tool.ipynb文件

import pymysql
from langchain.tools import tool
from pydantic import BaseModel,Field

# 1.定义数据库连接
def get_db_connection():
    return pymysql.connect(
        host='localhost',
        user='root',
        password='<密码>',
        database='<数据库名>', 
        charset='utf8'
    )

# 2.定义参数结构和描述，给大模型查看
class ScoreQueryInput(BaseModel):
    name: str = Field(description="学生姓名，必填")
    subject: str = Field(default=None,description="考试科目")
    exam_batch: str = Field(default=None,description="考试批次")

# 3. 定义Tool(核心!)
@tool
def query_student_score(name: str,subject: str=None,exam_batch: str=None):
    """
    根据学生姓名+科目/批次查询成绩，科目和考试批次至少提供一个。
    name：学生姓名
    subject：考试科目（可选）
    exam_batch：考试批次（可选）
    """
    # 条件至少满足其一
    if not subject and not exam_batch:
        return "请至少提供科目或考试批次"

    conn=get_db_connection()
    cursor=conn.cursor(pymysql.cursors.DictCursor)

    sql = """
    SELECT s.name,s.student_id,sc.subject,sc.exam_batch,sc.score
    FROM student s
    JOIN score sc ON s.id=sc.student_id
    WHERE s.name=%s
    """
    params=[name] #SQL语句中所需的参数列表

    #逻辑
    if subject:
        sql+="AND sc.subject=%s"
        params.append(subject)
    if exam_batch:
        sql+="AND sc.exam_batch=%s"
        params.append(exam_batch)

    cursor.execute(sql,params)
    results=cursor.fetchall()
    conn.close() #关闭连接

    if not results:
        return f"未找到【{name}】的成绩"
    return str(results)

4.构建 Agent 自动调用

from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor #新版导入
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 1. 初始化LLM
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    api_key="sk-evcXSBiYi5CTf1PspCHTi4Vo2QwDy18uF0d5wH503otiQ4hP",
    base_url="https://yinli.one/v1",
)

# 2.Agent
tools = [query_student_score]

# 提示词（固定写法）
prompt = ChatPromptTemplate.from_messages([
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 创建新版 agent
agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 3. 测试
result1 = agent_executor.invoke({"input": "张三 2026-01 的成绩"})
print(result1["output"])

result2 = agent_executor.invoke({"input": "李四的英语成绩"})
print(result2["output"])

result3 = agent_executor.invoke({"input": "张三 2026-02 的数学成绩"})
print(result3["output"])

5.连接数据库并测试连接

6.测试运行

得到类似结果则表明我们已经成功

> Entering new AgentExecutor chain...

Invoking: `query_student_score` with `{'name': '张三', 'exam_batch': '2026-01'}`

[{'name': '张三', 'student_id': '2026001', 'subject': '数学', 'exam_batch': '2026-01', 'score': 90}, {'name': '张三', 'student_id': '2026001', 'subject': '语文', 'exam_batch': '2026-01', 'score': 85}]张三在2026年1月的成绩如下：

- 数学：90分
- 语文：85分

> Finished chain.
张三在2026年1月的成绩如下：

- 数学：90分
- 语文：85分

> Entering new AgentExecutor chain...

Invoking: `query_student_score` with `{'name': '李四', 'subject': '英语'}`

[{'name': '李四', 'student_id': '2026002', 'subject': '英语', 'exam_batch': '2026-01', 'score': 92}, {'name': '李四', 'student_id': '2026002', 'subject': '英语', 'exam_batch': '2026-02', 'score': 96}]李四的英语成绩如下：

- 2026年1月考试：92分
- 2026年2月考试：96分

> Finished chain.
李四的英语成绩如下：
- 2026年1月考试：92分
- 2026年2月考试：96分

> Entering new AgentExecutor chain...

Invoking: `query_student_score` with `{'name': '张三', 'subject': '数学', 'exam_batch': '2026-02'}`

[{'name': '张三', 'student_id': '2026001', 'subject': '数学', 'exam_batch': '2026-02', 'score': 96}]张三在2026年2月的数学成绩是96分。

> Finished chain.
张三在2026年2月的数学成绩是96分。

🦆执行流程图

1. 导入库
   ↓
2. 定义数据库连接函数 get_db_connection()
   ↓
3. 定义参数结构 ScoreQueryInput（给LLM看参数说明）
   ↓
4. 定义 Tool：query_student_score
   ↓
5. 初始化大模型 llm = ChatOpenAI(...)
   ↓
6. 准备工具列表 tools = [query_student_score]
   ↓
7. 构建提示词模板 prompt
   包含 {input} + agent_scratchpad（占位）
   ↓
8. 创建 agent = create_openai_tools_agent(llm, tools, prompt)
   ↓
9. 创建执行器 agent_executor = AgentExecutor(...)
   ↓
=============================================
↓ 开始执行第一次查询
10. agent_executor.invoke({"input":"张三 2026-01 的成绩"})
   ↓
    【Agent 内部自动走下面流程】
    a. LLM 解析问题 → 决定调用 query_student_score
    b. 传入参数 name="张三", exam_batch="2026-01"
    c. 进入 Tool 内部执行：
       ① 校验参数
       ② conn = get_db_connection() 连接数据库
       ③ cursor = conn.cursor(DictCursor) 创建游标
       ④ 拼接 SQL
       ⑤ cursor.execute(sql, params) 执行查询
       ⑥ results = cursor.fetchall() 取全部数据
       ⑦ 关闭连接 conn.close()
       ⑧ 返回结果给 Agent
    d. LLM 拿到结果 → 整理成自然语言回答
   ↓
11. 打印 result1["output"]
=============================================
↓
12. 执行第二次 invoke：李四的英语成绩
    重复上述 Tool 调用流程
↓
13. 打印 result2["output"]
↓
14. 执行第三次 invoke：张三 2026-02 数学成绩
    重复流程
↓
15. 打印 result3["output"]