LangChain 0.3手写AI Agent：工具调用与记忆管理全拆解

python 3.11.7
langchain 0.3.7
langchain-openai 0.2.9
openai 1.54.3
pydantic 2.9.2

from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing import Literal, Optional

class WeatherInput(BaseModel):
    city: str = Field(description="城市名，中文，如'北京'")
    date: Optional[str] = Field(default="today", description="日期，格式YYYY-MM-DD，默认today")
    unit: Literal["celsius", "fahrenheit"] = Field(default="celsius", description="温度单位")

@tool(args_schema=WeatherInput)
def get_weather(city: str, date: str = "today", unit: str = "celsius") -> dict:
    """查询指定城市天气，返回最高/最低温度"""
    # 模拟真实API调用，生产环境替换成requests.get(...)
    mock_db = {
        ("北京", "today"): {"high": 15, "low": 5},
        ("上海", "today"): {"high": 18, "low": 10},
    }
    key = (city, date)
    if key not in mock_db:
        raise ValueError(f"没有{city}在{date}的天气数据")  # 故意抛错验证错误处理
    high = mock_db[key]["high"]
    low = mock_db[key]["low"]
    if unit == "fahrenheit":
        high = round(high * 9/5 + 32, 1)
        low = round(low * 9/5 + 32, 1)
    return {"city": city, "high": high, "low": low, "unit": unit}

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage
import time, json

class SimpleAgent:
    def __init__(self, tools, model_name="gpt-4o-mini", max_iters=5):
        self.tools = {t.name: t for t in tools}
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=0.2,
            max_tokens=1024
        ).bind_tools(list(self.tools.values()))  # 关键：绑定schema到模型
        self.max_iters = max_iters
        self.memory = []  # 简化版记忆：list存储消息
        self.error_streak = 0

    def _run_tool(self, tool_name: str, tool_args: dict):
        """执行工具，带重试与异常捕获"""
        try:
            tool_func = self.tools[tool_name]
            result = tool_func.invoke(tool_args)  # LangChain会自动校验参数
            self.error_streak = 0  # 成功则重置连续错误计数
            return result
        except Exception as e:
            self.error_streak += 1
            delay = min(2 ** self.error_streak, 10)  # 指数退避：1s,2s,4s...
            print(f"⚠️ 工具{tool_name}执行失败: {str(e)}，{delay}秒后重试")
            time.sleep(delay)
            # 返回错误信息作为消息，让模型自己决策
            return {"error": f"工具调用失败: {str(e)}，请检查参数或换一种方式"}

    def run(self, user_input: str) -> str:
        # 初始化消息：系统提示 + 记忆 + 新输入
        messages = [SystemMessage(content="你是运维助手。需要信息时调用工具，结果不满足要求时主动重试。")]
        # 注入记忆（最近3轮）
        messages.extend(self.memory[-6:])  # 每轮2条消息(AI+User)，取最近3轮
        messages.append(HumanMessage(content=user_input))

        for i in range(self.max_iters):
            print(f"🔄 第{i+1}次迭代...")
            response = self.llm.invoke(messages)
            messages.append(response)

            # 检查是否有工具调用请求
            if not response.tool_calls:
                # 没有工具调用，直接返回最终答案
                self.memory.extend([HumanMessage(content=user_input), response])
                self.memory = self.memory[-20:]  # 裁剪记忆最多20条消息
                return response.content
            else:
                # 执行所有工具调用（可能并行多个）
                for tc in response.tool_calls:
                    print(f"🔧 调用工具: {tc['name']} 参数: {tc['args']}")
                    tool_result = self._run_tool(tc["name"], tc["args"])
                    # 将工具结果包装成ToolMessage追加
                    tool_msg = ToolMessage(
                        content=json.dumps(tool_result, ensure_ascii=False),
                        tool_call_id=tc["id"]  # 必须关联id，否则模型无法关联
                    )
                    messages.append(tool_msg)
        # 超迭代次数
        fallback = "已达最大迭代次数，无法得出结论。请尝试更具体地描述需求。"
        self.memory.append(HumanMessage(content=user_input))
        self.memory.append(AIMessage(content=fallback))
        return fallback

# 实例化并运行
agent = SimpleAgent(tools=[get_weather])
result = agent.run("北京今天最高温和最低温差多少度？")
print("🤖 最终回答:", result)

🔄 第1次迭代...
🔧 调用工具: get_weather 参数: {'city': '北京', 'date': 'today', 'unit': 'celsius'}
🔄 第2次迭代...
🤖 最终回答: 北京今天最高温度15°C，最低温度5°C，温差为10°C。