MySQL、Memory 到 Agent 整合 · 第 35—53 课完整笔记

Python 基础
→ HTTP / API
→ FastAPI
→ 调用 Qwen
→ messages 多轮对话
→ conversation_id 区分会话
→ memory.py 用 Python dict 保存聊天历史

Client → FastAPI → memory.py / dict → model.py → Qwen
                    ↑                         ↓
                    └──── 保存新对话 ─────────┘

Client → FastAPI
           ├─ memory.py → MySQL → 读取历史
           ├─ model.py → Qwen → 得到回答
           └─ memory.py → MySQL → 保存本轮 user + assistant

tool_call_test.py
→ Qwen 读取工具说明
→ 产生 tool_calls
→ Python 执行 calculator
→ 把结果回传 Qwen
→ 最终回答

tool_call_test.py
→ run_agent(messages)
→ 循环调用 Qwen
→ 查询 Tool Registry
→ 执行工具
→ 添加 Observation
→ 继续循环 / 返回最终答案

Client / Swagger
       ↓
main.py：接收 conversation_id + message
       ↓
memory.py → MySQL：读取历史
       ↓
历史 + 当前 user message
       ↓
agent.py：run_agent()
       ├─ Qwen：决定下一步
       ├─ tools.py：Schema + Registry + calculator
       └─ 工具结果 → 上下文 → 再次调用 Qwen
       ↓
最终 answer
       ↓
memory.py → MySQL：只保存当前 user + 最终 assistant
       ↓
FastAPI 返回 JSON

memory = {
    "abc123": [
        {"role": "user", "content": "我的名字叫小明"},
        {"role": "assistant", "content": "好的，我记住了"}
    ]
}

启动 Uvicorn
→ 加载 Python 模块
→ 创建 memory 字典
→ 对话不断追加到字典
→ 停止这个 Python 进程
→ 这份进程内存不再保留
→ 再启动时重新创建初始 memory

get_messages → dict.get
add_message  → list.append

get_messages → SELECT
add_message  → INSERT + commit

history = get_messages(conversation_id)
add_message(conversation_id, "user", question)

MySQL 保存历史
→ Python 根据 conversation_id 查询
→ 把结果整理成 messages
→ 本次请求把 messages 交给 Qwen
→ Qwen 根据本次收到的上下文回答

Chat History → RAM
Long-term Memory → 数据库

user：我现在在学习 FastAPI。
assistant：可以先从路由和请求体开始。
user：我想以后做 Agent。
assistant：接下来可以学习工具调用。

用户正在学习 AI Agent 开发。
用户希望通过实践理解 Python 与后端。

存储维度：
临时 RAM / 持久化数据库

信息维度：
聊天原文 / 提炼信息 / 执行轨迹

conversations
└─ abc123
   ├─ user：我的名字叫岭西
   ├─ assistant：你好
   ├─ user：我叫什么名字？
   └─ assistant：你叫岭西

一个 User
→ 多个 Conversation
→ 每个 Conversation 又包含多个 Message

conversations.user_id → users.id
messages.conversation_id → conversations.conversation_id

CREATE TABLE conversations (
    id INT AUTO_INCREMENT PRIMARY KEY,
    conversation_id VARCHAR(100) NOT NULL UNIQUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

id = 1
conversation_id = abc123

CREATE TABLE messages (
    id INT AUTO_INCREMENT PRIMARY KEY,
    conversation_id VARCHAR(100) NOT NULL,
    role VARCHAR(20) NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

SELECT role, content
FROM messages
WHERE conversation_id = 'abc123'
ORDER BY id ASC;

mysql -u root -p

5.5.11

SELECT VERSION();
SHOW DATABASES;
CREATE DATABASE agent_demo CHARACTER SET utf8mb4;
USE agent_demo;
SELECT DATABASE();

SHOW TABLES;
DESC conversations;
DESC messages;

INSERT INTO conversations (conversation_id)
VALUES ('abc123');

INSERT INTO messages (conversation_id, role, content)
VALUES ('abc123', 'user', '我的名字叫岭西');

INSERT INTO messages (conversation_id, role, content)
VALUES ('abc123', 'assistant', '你好');

SELECT role, content
FROM messages
WHERE conversation_id = 'abc123'
ORDER BY id ASC;

表结构正确
→ SQL 能执行
→ 内容能保存
→ 内容能查回

mysql> SHOW DATABASES
    ->

\c

ALTER TABLE conversations
CHANGE conversion_id conversation_id VARCHAR(100) NOT NULL;

DESC conversations;

Python 业务代码
→ PyMySQL
→ MySQL 协议
→ MySQL Server
→ SQL 结果
→ Python 数据对象

python -m pip install pymysql

DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=root
DB_PASSWORD=填写本机数据库密码
DB_NAME=agent_demo

# file: config.py
import os
from pathlib import Path

from dotenv import load_dotenv

load_dotenv(Path(__file__).resolve().parent / ".env")

API_KEY = os.getenv("API_KEY")
BASE_URL = os.getenv("BASE_URL")
DEFAULT_MODEL = os.getenv("DEFAULT_MODEL")

DB_HOST = os.getenv("DB_HOST", "127.0.0.1")
DB_PORT = int(os.getenv("DB_PORT", "3306"))
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_NAME = os.getenv("DB_NAME", "agent_demo")

DB_PORT = int(os.getenv("DB_PORT", "3306"))

# file: db.py
import pymysql

from config import (
    DB_HOST,
    DB_PORT,
    DB_USER,
    DB_PASSWORD,
    DB_NAME,
)


def get_connection():
    return pymysql.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME,
        charset="utf8mb4",
    )

# file: test_db.py
from db import get_connection

connection = get_connection()
try:
    with connection.cursor() as cursor:
        cursor.execute("SELECT VERSION();")
        result = cursor.fetchone()
        print("MySQL版本：", result)
finally:
    connection.close()

MySQL版本： ('5.5.11',)

读取配置
→ 建立连接
→ 通过身份验证
→ 执行 SELECT VERSION()
→ 取出一行
→ 返回 Python

cursor.execute("SELECT role, content FROM messages;")
rows = cursor.fetchall()

("user", "我的名字叫岭西")

(
    ("user", "我的名字叫岭西"),
    ("assistant", "你好"),
)

谁说的 + 说了什么

[
    {"role": "user", "content": "我的名字叫岭西"},
    {"role": "assistant", "content": "你好"},
]

cursor.execute(
    """
    SELECT role, content
    FROM messages
    WHERE conversation_id = %s
    ORDER BY id ASC;
    """,
    (conversation_id,),
)

value = "abc123"
a = (value)     # str，括号只是分组
b = (value,)    # tuple，只有一个元素

execute INSERT
→ 尝试执行写入

commit
→ 提交事务

close
→ 关闭连接，不应拿它代替 commit

get_messages(conversation_id)
→ 读这一段会话的历史

ensure_conversation(conversation_id)
→ 没有这段会话时创建它

add_message(conversation_id, role, content)
→ 先确保会话存在，再保存一条消息

接收 conversation_id
→ 建立数据库连接
→ SELECT role, content
→ WHERE 限定会话
→ ORDER BY 保证顺序
→ fetchall 取回行
→ tuple 行转为 dict
→ 组成 list
→ 关闭资源
→ return messages

messages = []
for row in rows:
    messages.append({
        "role": row[0],
        "content": row[1],
    })

new999 | user | 这是一个新的会话

SELECT id WHERE conversation_id = new999
→ fetchone()

查到一行：
    不再重复 INSERT

返回 None：
    INSERT conversations
    commit

# file: memory.py
from typing import Dict, List

from db import get_connection


def ensure_conversation(conversation_id: str) -> None:
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT id
                FROM conversations
                WHERE conversation_id = %s;
                """,
                (conversation_id,),
            )
            result = cursor.fetchone()

            if result is None:
                cursor.execute(
                    """
                    INSERT INTO conversations (conversation_id)
                    VALUES (%s);
                    """,
                    (conversation_id,),
                )
                connection.commit()
    except Exception:
        connection.rollback()
        raise
    finally:
        connection.close()


def get_messages(conversation_id: str) -> List[Dict[str, str]]:
    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT role, content
                FROM messages
                WHERE conversation_id = %s
                ORDER BY id ASC;
                """,
                (conversation_id,),
            )
            rows = cursor.fetchall()

            messages = []
            for row in rows:
                messages.append({
                    "role": row[0],
                    "content": row[1],
                })

        return messages
    finally:
        connection.close()


def add_message(
    conversation_id: str,
    role: str,
    content: str,
) -> None:
    ensure_conversation(conversation_id)

    connection = get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO messages (
                    conversation_id,
                    role,
                    content
                )
                VALUES (%s, %s, %s);
                """,
                (conversation_id, role, content),
            )
        connection.commit()
    except Exception:
        connection.rollback()
        raise
    finally:
        connection.close()

main.py → get_messages → dict

main.py → get_messages → PyMySQL → MySQL

from memory import get_messages

messages = get_messages("abc123")
print(messages)

[
    {"role": "user", "content": "我的名字叫岭西"},
    {"role": "assistant", "content": "你好"},
]

from memory import add_message

add_message("new999", "user", "这是一个新的会话")
print("保存完成")

SELECT *
FROM conversations
WHERE conversation_id = 'new999';

SELECT role, content
FROM messages
WHERE conversation_id = 'new999'
ORDER BY id ASC;

conversations：
new999 仍然只有一个会话记录

messages：
每执行一次 add_message，都会新增一条消息

收到 conversation_id + question
→ get_messages 读取历史
→ 加入当前问题
→ call_model 调用 Qwen
→ 得到回答
→ add_message 保存 user
→ add_message 保存 assistant
→ 返回客户端

from typing import Optional

from pydantic import BaseModel


class ChatRequest(BaseModel):
    conversation_id: str
    question: str
    model: Optional[str] = None

history = get_messages(request.conversation_id)

messages = []
messages.extend(history)
messages.append({
    "role": "user",
    "content": request.question,
})

answer = call_model(messages=messages, model=request.model)

add_message(request.conversation_id, "user", request.question)
add_message(request.conversation_id, "assistant", answer)

[user, assistant, user, assistant]

[[user, assistant, user, assistant]]

{
  "conversation_id": "persist001",
  "question": "我的名字叫岭西，请记住"
}

SELECT role, content
FROM messages
WHERE conversation_id = 'persist001'
ORDER BY id ASC;

{
  "conversation_id": "persist001",
  "question": "我叫什么名字？"
}

确认数据库已有历史
→ 停止 Uvicorn
→ 重新启动 Uvicorn
→ 用 persist001 再问名字
→ 检查仍然能回答

理解当前问题
→ 参考历史
→ 生成文本回答

LLM：理解任务和决定下一步
Tools：提供真实能力
Python Runtime：调度并执行
Memory：提供已有背景
Loop：让决策和执行反复衔接

程序员：
定义函数、参数规则、允许的工具

LLM：
选择这次是否调用、调用哪个工具、给出哪些参数值

Python：
查找函数、解析参数、真正执行

Tool：
完成具体计算或外部操作

LLM：
读取执行结果，再组织回答或提出下一次动作

def calculator(a: float, b: float, operation: str) -> float:
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        return a / b
    else:
        raise ValueError("不支持的运算类型")

result = calculator(12, 8, "multiply")
print(result)

a、b、operation 这些参数名与含义
→ 程序员定义

本次 a=12、b=8、operation=multiply
→ 模型根据用户问题选择

calculator_tool = {
    "type": "function",
    "function": {
        "name": "calculator",
        "description": "执行两个数字之间的基础数学运算",
        "parameters": {
            "type": "object",
            "properties": {
                "a": {
                    "type": "number",
                    "description": "第一个数字",
                },
                "b": {
                    "type": "number",
                    "description": "第二个数字",
                },
                "operation": {
                    "type": "string",
                    "description": "运算类型",
                    "enum": [
                        "add",
                        "subtract",
                        "multiply",
                        "divide",
                    ],
                },
            },
            "required": ["a", "b", "operation"],
        },
    },
}

completion = client.chat.completions.create(
    model=DEFAULT_MODEL,
    messages=messages,
    tools=[calculator_tool],
)

ChatCompletionMessage(
    role="assistant",
    content="",
    tool_calls=[
        ToolCall(
            id="call_...",
            function={
                name: "calculator",
                arguments: '{"a": 12, "b": 8, "operation": "multiply"}'
            }
        )
    ]
)

请使用 calculator
参数是 12、8、multiply

message = completion.choices[0].message
tool_call = message.tool_calls[0]
tool_name = tool_call.function.name
arguments_text = tool_call.function.arguments

completion
└─ choices 列表
   └─ 第 0 个候选
      └─ message
         ├─ content
         └─ tool_calls 列表
            └─ 某一次 tool_call
               ├─ id
               └─ function
                  ├─ name
                  └─ arguments

arguments_text = '{"a": 12, "b": 8, "operation": "multiply"}'

import json

arguments = json.loads(arguments_text)

{
    "a": 12,
    "b": 8,
    "operation": "multiply",
}

result = calculator(
    a=arguments["a"],
    b=arguments["b"],
    operation=arguments["operation"],
)

模型选择工具： calculator
模型生成参数： {'a': 12, 'b': 8, 'operation': 'multiply'}
工具执行结果： 96

模型选工具
→ 生成参数
→ Python 解析
→ 真正执行函数

print(result)

messages.append(message)

messages.append({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "content": str(result),
})

name = calculator
→ 调用哪个工具

id = call_aaa
→ 本轮中的哪一次具体调用

call_a → calculator(12, 8, multiply) → 96
call_b → calculator(96, 4, divide)  → 24.0

[
  {
    "role": "user",
    "content": "请使用计算器计算12乘以8"
  },
  {
    "role": "assistant",
    "content": "",
    "tool_calls": [
      {
        "id": "call_example",
        "type": "function",
        "function": {
          "name": "calculator",
          "arguments": "{\"a\":12,\"b\":8,\"operation\":\"multiply\"}"
        }
      }
    ]
  },
  {
    "role": "tool",
    "tool_call_id": "call_example",
    "content": "96"
  }
]

assistant：call_a + call_b
tool：call_a 的结果
tool：call_b 的结果
下一次 LLM 请求

第一次 LLM
→ 执行工具
→ 第二次 LLM
→ 结束

while True:
    completion = client.chat.completions.create(
        model=DEFAULT_MODEL,
        messages=messages,
        tools=tool_schemas,
    )
    message = completion.choices[0].message
    messages.append(message)

    if not message.tool_calls:
        print("最终回答：", message.content)
        break

    for tool_call in message.tool_calls:
        # 解析参数、执行工具、追加 tool message
        pass

while：
Agent 还要不要再次请求模型？

for：
这一次模型返回的多个 tool_calls，要逐个处理。

Step 1：
LLM 返回两个 Tool Call
for 执行两次

Step 2：
LLM 没有 Tool Call
直接返回最终答案

LLM：先提出 12 × 8
→ Python 算出 96
→ 把 96 给 LLM
→ LLM 再提出 96 ÷ 4
→ Python 算出 24.0
→ 把 24.0 给 LLM
→ 最终答案

Step 1：模型一次生成
    calculator(a=12, b=8, operation=multiply)
    calculator(a=96, b=4, operation=divide)

Python 用 for 顺序执行，返回 96 与 24.0

Step 2：模型返回最终答案

if tool_name == "calculator":
    result = calculator(
        a=arguments["a"],
        b=arguments["b"],
        operation=arguments["operation"],
    )

tool_registry = {
    "calculator": calculator,
}

"calculator": calculator

calculator()

tool_function = tool_registry.get(tool_name)

tool_function → calculator 函数对象

tool_function → None

tool_registry[tool_name]

arguments = {
    "a": 12,
    "b": 8,
    "operation": "multiply",
}

result = tool_function(**arguments)

result = tool_function(a=12, b=8, operation="multiply")

key → 参数名
value → 参数值

Schema 中 name = calculator
        ↓
模型输出 name = calculator
        ↓
Registry 中 key = calculator
        ↓
calculator 函数

LLM 请求
→ 工具调用
→ LLM 请求
→ 工具调用
→ 持续重复

step = 0
while step < max_steps:
    step += 1
    # 当前一轮处理

finished = False

if step >= max_steps:
    print("达到最大执行步数")

循环内发现最终回答 → return 最终回答
循环自然耗尽       → return 超限提示

tool_function = tool_registry.get(tool_name)

if tool_function is None:
    result = "工具不存在：" + tool_name
else:
    result = tool_function(**arguments)

try:
    result = tool_function(**arguments)
except Exception as e:
    result = "工具执行失败：" + str(e)

calculator(a=10, b=0, operation="divide")

工具执行失败：division by zero

arguments = json.loads(tool_call.function.arguments)

try:
    result = tool_function(**arguments)
except Exception as e:
    result = "工具执行失败：" + str(e)

无法解析 JSON
参数不是对象
工具名不存在
参数不匹配
工具内部执行失败

配置模型客户端
定义工具 Schema
管理工具 Registry
准备问题
运行 Agent Loop
打印结果

agent-demo/
├─ .env
├─ .gitignore
├─ requirements.txt
├─ config.py
├─ db.py
├─ memory.py
├─ model.py
├─ tools.py
├─ agent.py
├─ main.py
├─ test_db.py
├─ linshitest.py
├─ conversationtest.py
└─ tool_call_test.py

# file: tools.py
def calculator(
    a: float,
    b: float,
    operation: str,
) -> float:
    if operation == "add":
        return a + b
    elif operation == "subtract":
        return a - b
    elif operation == "multiply":
        return a * b
    elif operation == "divide":
        return a / b
    else:
        raise ValueError("不支持的运算类型")


calculator_tool = {
    "type": "function",
    "function": {
        "name": "calculator",
        "description": "执行两个数字之间的基础数学运算",
        "parameters": {
            "type": "object",
            "properties": {
                "a": {
                    "type": "number",
                    "description": "第一个数字",
                },
                "b": {
                    "type": "number",
                    "description": "第二个数字",
                },
                "operation": {
                    "type": "string",
                    "description": "运算类型",
                    "enum": [
                        "add",
                        "subtract",
                        "multiply",
                        "divide",
                    ],
                },
            },
            "required": ["a", "b", "operation"],
        },
    },
}

tool_schemas = [calculator_tool]

tool_registry = {
    "calculator": calculator,
}

# file: agent.py
import json

from openai import OpenAI

from config import API_KEY, BASE_URL, DEFAULT_MODEL
from tools import tool_registry, tool_schemas


client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
)


def run_agent(messages, max_steps: int = 10):
    step = 0

    while step < max_steps:
        step += 1
        print("当前Step:", step)

        completion = client.chat.completions.create(
            model=DEFAULT_MODEL,
            messages=messages,
            tools=tool_schemas,
        )
        message = completion.choices[0].message

        print("模型 content：", message.content)
        print("模型 tool_calls：", message.tool_calls)

        # 保留模型的调用请求，后面的 tool result 才有对应关系。
        messages.append(message)

        if not message.tool_calls:
            return message.content

        for tool_call in message.tool_calls:
            tool_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)

            print("模型选择工具：", tool_name)
            print("模型生成参数：", arguments)

            tool_function = tool_registry.get(tool_name)

            if tool_function is None:
                result = "工具不存在：" + tool_name
            else:
                try:
                    result = tool_function(**arguments)
                except Exception as e:
                    result = "工具执行失败：" + str(e)

            print("工具执行结果：", result)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result),
            })

    # 必须与 while 同一级，表示循环耗尽后才执行。
    return "Agent达到最大执行步数，任务未完成"

answer = run_agent(messages)
print("最终回答：", answer)

answer = run_agent(messages)

messages.append({
    "role": "tool",
    "tool_call_id": tool_call.id,
    "content": str(result),
})

# file: tool_call_test.py
from agent import run_agent


if __name__ == "__main__":
    messages = [
        {
            "role": "user",
            "content": "请使用计算器，先计算12乘以8，再把结果除以4。",
        }
    ]

    answer = run_agent(messages)
    print("最终回答：", answer)

python tool_call_test.py

历史 + 当前问题 → call_model → answer

历史 + 当前问题 → run_agent → answer

# file: main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from agent import run_agent
from memory import get_messages, add_message


app = FastAPI()


class ChatRequest(BaseModel):
    conversation_id: str
    message: str


@app.post("/chat")
def chat(request: ChatRequest):
    try:
        messages = get_messages(request.conversation_id)

        messages.append({
            "role": "user",
            "content": request.message,
        })

        answer = run_agent(messages)

        add_message(
            request.conversation_id,
            "user",
            request.message,
        )
        add_message(
            request.conversation_id,
            "assistant",
            answer,
        )

        return {"answer": answer}

    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=str(e),
        )

{
  "conversation_id": "agent001",
  "message": "请使用计算器，先计算12乘以8，再把结果除以4。"
}

{
  "conversation_id": "agent001",
  "question": "请计算12乘以8"
}

user：我的名字叫岭西
assistant：你好，岭西

请计算12乘以8，再把结果除以4

messages = get_messages(conversation_id)

messages：
旧 user
旧 assistant

messages：
旧 user
旧 assistant
本次 user

messages：
旧 user
旧 assistant
本次 user
本轮 assistant(tool_calls)
本轮 tool result
本轮 tool result
最终 assistant

answer = 最终结果是24.0

INSERT 当前 user
INSERT 最终 assistant

旧 user
旧 assistant
本次 user
本次最终 assistant

for message in messages:
    全部写入 MySQL

12 × 8 = 96
96 ÷ 4 = 24.0
最终结果 24.0

是否新增 role=tool 的独立记录？
是否保存 assistant 的结构化 tool_calls？
是否保存了内部 tool_call_id 等字段？

run_agent 超限
→ 返回“任务未完成”字符串
→ 路由没有收到异常
→ 仍可能保存该字符串并返回 200

user：
请先计算12乘以8，再把结果除以4。

assistant：
最终结果是24.0。

user：
刚才最终算出来是多少？

assistant：
刚才最终算出来是24.0。

开始运行
→ Step 1
→ 模型提出 calculator 调用
→ arguments = 12、8、multiply
→ 工具返回 96
→ 第二个 calculator 调用
→ arguments = 96、4、divide
→ 工具返回 24.0
→ Step 2
→ 模型返回最终回答
→ 运行结束

从 MySQL 读取的 Chat History
+
当前 user message
+
当前 assistant tool_calls
+
当前 tool results
+
最终 assistant answer

模型选对工具了吗？
→ name

参数正确吗？
→ arguments

工具真执行了吗？
→ 工具日志 / 返回值

工具自己报错了吗？
→ exception / error result

结果是否回传了？
→ role=tool + tool_call_id

模型是否再次读取了？
→ 下一轮 Step / 模型调用

最后是否提前结束？
→ return / break 所在缩进

文件能否找到？
→ 模块能否导入？
→ 配置能否读到？
→ 网络能否连接？
→ SQL / 模型请求能否执行？
→ 返回结构能否解析？
→ 工具能否执行？
→ 结果能否回传？
→ 循环能否继续？
→ 数据能否保存？
→ 界面能否正确显示？

SHOW DATABASES
->

SHOW TABLES;
DESC conversations;
DESC messages;

VALUES ('abc123'， 'assistant'， '你好')

VALUES ('abc123', 'assistant', '你好');

VALUES ('abc123', 'assistant', '好的，我记住了');

ERROR 1046 (3D000): No database selected

SELECT DATABASE();
USE agent_demo;
SELECT DATABASE();

ERROR 1366 (HY000): Incorrect string value:
'\xCE\xD2\xB5\xC4...'
for column 'content'

Windows 中文 CMD 可能使用 GBK / CP936
→ 客户端发送相应编码的字节
→ 连接却按其他字符集解释
→ 写入转换失败

SET NAMES gbk;

charset="utf8mb4"

Python / PyMySQL
→ utf8mb4 连接
→ MySQL 中正确保存中文

MySQL 命令行
→ 返回 UTF-8 字节
→ CMD 按另一种编码解释
→ 显示成乱码

from memory import get_messages

print(get_messages("new999"))

SHOW VARIABLES LIKE 'character_set%';

SHOW CREATE TABLE messages;
SHOW FULL COLUMNS FROM messages;

exit

chcp 65001
mysql --default-character-set=utf8mb4 -u root -p

USE agent_demo;

SELECT role, content
FROM messages
WHERE conversation_id = 'new999'
ORDER BY id ASC;

SET NAMES utf8mb4;

MySQL 连接设置正确
但终端显示仍不匹配

python linshitest

linshitest.py

python linshitest.py

from tools import calculator
ModuleNotFoundError: No module named 'tools'

openai.APIConnectionError: Connection error.
httpx.ConnectError:
EOF occurred in violation of protocol (_ssl.c:1125)

httpcore\_sync\http_proxy.py

BASE_URL 是否为之前已跑通的服务地址
→ 代理或 VPN 是否影响这条连接
→ HTTP_PROXY / HTTPS_PROXY / ALL_PROXY 等环境配置
→ 网络是否可达
→ 同一客户端的最小普通模型请求是否成功

最终回答：Agent达到最大执行步数，任务未完成

当前Step: 1
工具执行结果：96
工具执行结果：24.0
最终回答：Agent达到最大执行步数，任务未完成

def wrong_loop_example():
    step = 0
    while step < 10:
        step += 1
        # 本轮处理……
        return "Agent达到最大执行步数，任务未完成"

def correct_loop_shape():
    step = 0
    while step < 10:
        step += 1
        # 本轮处理；正常完成时在这里 return 答案。

    return "Agent达到最大执行步数，任务未完成"

当前Step: 1
模型 content：
模型 tool_calls：[两个 calculator 调用]

模型选择工具：calculator
模型生成参数：{'a': 12, 'b': 8, 'operation': 'multiply'}
工具执行结果：96

模型选择工具：calculator
模型生成参数：{'a': 96, 'b': 4, 'operation': 'divide'}
工具执行结果：24.0

当前Step: 2
模型 content：最终答案是 24。
模型 tool_calls：None
最终回答：最终答案是 24。

$ 12 \times 8 = 96 $
$ 96 \div 4 = 24.0 $
**24.0**

{
  "conversation_id": "agent001",
  "message": "请使用计算机，先计算12乘以8，再把结果除以4。"
}

200

12 × 8 = 96
96 ÷ 4 = 24.0
最终结果是 24.0

SELECT role, content
FROM messages
WHERE conversation_id = 'agent001'
ORDER BY id ASC;

role
----
user
assistant

2 rows in set

{
  "conversation_id": "agent001",
  "message": "刚才最终算出来是多少？"
}

{
  "answer": "刚才最终算出来是 **24.0**。"
}

数据库可登录
→ 手工 SQL
→ Python 数据库连接
→ get_messages
→ add_message / ensure_conversation
→ 单独 calculator
→ 模型提出 Tool Call
→ Python 解析并执行
→ Tool Result 回传
→ Agent Loop
→ 模块化脚本
→ FastAPI 整合
→ MySQL 持久化与后续追问

USE agent_demo;
SHOW TABLES;
DESC conversations;
DESC messages;

SELECT role, content
FROM messages
WHERE conversation_id = 'new999'
ORDER BY id ASC;

python test_db.py
python linshitest.py

python tool_call_test.py

至少一次模型返回 tool_calls
→ 函数实际执行
→ 对应 tool result 被追加
→ 后续模型调用
→ 没有 tool_calls 的最终回答

uvicorn main:app --reload

Application startup complete.

{
  "conversation_id": "review_agent_001",
  "message": "请使用计算器，先计算12乘以8，再把结果除以4。"
}

SELECT role, content
FROM messages
WHERE conversation_id = 'review_agent_001'
ORDER BY id ASC;

{
  "conversation_id": "review_agent_001",
  "message": "刚才最终算出来是多少？"
}

客户端发送 conversation_id + message
                  ↓
FastAPI 校验请求
                  ↓
get_messages(conversation_id)
                  ↓
MySQL SELECT → 按顺序恢复历史
                  ↓
追加当前 user message
                  ↓
run_agent(messages)
                  ↓
Qwen 读取上下文与 Tool Schema
                  ↓
        有 tool_calls 吗？
          ├─ 否 → 返回最终文本
          └─ 是
              ↓
          解析 JSON arguments
              ↓
          Registry 查找函数
              ↓
          function(**arguments)
              ↓
          成功结果或错误 Observation
              ↓
          追加对应 tool message
              ↓
          回到 Qwen，受 max_steps 限制
                  ↓
main.py 得到 answer
                  ↓
只 INSERT 本轮 user + 最终 assistant
                  ↓
FastAPI 返回 JSON

MySQL 持久化 Chat History
+
FastAPI 请求入口
+
Qwen 工具调用
+
Python 工具执行
+
Tool Registry
+
多轮 Agent Loop
+
轮次上限与基础工具异常处理
+
模块化组织