手把手教你用 Python 从零写一个自己的 Agent(带记忆+工具调用)

# ================= 初始化 MySQL 表结构 =================def init_db():    """确保 conversation_messages 表存在"""    conn = pymysql.connect(**DB_CONFIG)    cursor = conn.cursor()    cursor.execute("""        CREATE TABLE IF NOT EXISTS conversation_messages (            id INT AUTO_INCREMENT PRIMARY KEY,            session_id VARCHAR(64) NOT NULL,            message_data JSON NOT NULL,            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,            INDEX idx_session_id (session_id)        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4    """)    conn.commit()    cursor.close()    conn.close()

# ================= 保存单条消息到数据库 =================def save_message(session_id: str, message_dict: dict):    """将一条消息（dict格式）存入数据库"""    conn = pymysql.connect(**DB_CONFIG)    cursor = conn.cursor()    json_str = json.dumps(message_dict, ensure_ascii=False)    cursor.execute(        "INSERT INTO conversation_messages (session_id, message_data) VALUES (%s, %s)",        (session_id, json_str)    )    conn.commit()    cursor.close()    conn.close()# ================= 加载指定会话的所有历史消息 =================def load_history(session_id: str):    """返回该会话的消息列表（按时间升序）"""    conn = pymysql.connect(**DB_CONFIG)    cursor = conn.cursor()    cursor.execute(        "SELECT message_data FROM conversation_messages WHERE session_id = %s ORDER BY id ASC",        (session_id,)    )    rows = cursor.fetchall()    cursor.close()    conn.close()    messages = [json.loads(row[0]) for row in rows]    return messages# ================= 获取所有已存在的会话ID =================def list_sessions():    """返回数据库中的所有 session_id（去重）"""    conn = pymysql.connect(**DB_CONFIG)    cursor = conn.cursor()    cursor.execute("SELECT DISTINCT session_id FROM conversation_messages ORDER BY session_id")    sessions = [row[0] for row in cursor.fetchall()]    cursor.close()    conn.close()    return sessions

# ================= DeepSeek 客户端配置 =================client = OpenAI(    api_key="sk-b0fa1ae8668449a4815b01d268419758",  # 请替换为你的真实 API Key    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")

#================= 工具函数（天气、计算） =================def get_weather(city: str) -> str:    weather_db = {"北京": "晴天，30度", "上海": "小雨，22度", "广州": "雷阵雨，30度"}    return weather_db.get(city, f"未找到{city}的天气")def calculate(expression: str) -> str:    try:        result = eval(expression)        return f"计算结果: {result}"    except:        return "表达式错误"

tools = [    {        "type": "function",        "function": {            "name": "get_weather",            "description": "查询某个城市的天气",            "parameters": {                "type": "object",                "properties": {"city": {"type": "string", "description": "城市名"}},                "required": ["city"]            }        }    },    {        "type": "function",        "function": {            "name": "calculate",            "description": "计算数学表达式",            "parameters": {                "type": "object",                "properties": {"expression": {"type": "string", "description": "数学表达式"}},                "required": ["expression"]            }        }    }]

available_functions = {"get_weather": get_weather, "calculate": calculate}def run_agent(messages, session_id):    """    处理当前对话历史（内存中的 messages 列表），    调用模型、执行工具、保存新消息到数据库，并返回更新后的 messages。    """    response = client.chat.completions.create(        model="qwen-plus",        messages=messages,        tools=tools,        tool_choice="auto"    )    response_message = response.choices[0].message    tool_calls = response_message.tool_calls    # 保存模型的初始响应    msg_dict = response_message.model_dump()    messages.append(msg_dict)    save_message(session_id, msg_dict)    if not tool_calls:        print("Agent:", response_message.content)        return messages    # 执行工具调用    for tool_call in tool_calls:        func_name = tool_call.function.name        func_args = json.loads(tool_call.function.arguments)        result = available_functions[func_name](**func_args)        print(f"[工具调用] {func_name}({func_args}) -> {result}")        tool_msg = {            "role": "tool",            "tool_call_id": tool_call.id,            "content": result        }        messages.append(tool_msg)        save_message(session_id, tool_msg)    # 生成最终回复    final_response = client.chat.completions.create(        model="qwen-plus",        messages=messages    )    final_message = final_response.choices[0].message    final_dict = final_message.model_dump()    messages.append(final_dict)    save_message(session_id, final_dict)    print("Agent:", final_message.content)    return messages