LangChain 0.3手写Agent：工具注册与记忆回收机制实现

Python 3.11.8
langchain==0.3.1
langchain-core==0.3.15
langchain-openai==0.2.8
openai==1.55.3
redis==5.0.7  # 用于记忆外部存储，可选

User Input → AgentExecutor (循环控制)
                ├─ 工具注册表 (dict[str, BaseTool])
                ├─ 记忆管理器 (LRU缓存 + 时间衰减)
                └─ 错误处理中间件 (重试装饰器)

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type, Optional
import requests

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，中文")
    date: Optional[str] = Field(default="today", description="日期，格式YYYY-MM-DD")

class WeatherTool(BaseTool):
    name: str = "weather_query"
    description: str = "查询指定城市当日天气，返回温度和湿度。"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, date: str = "today", **kwargs) -> str:
        # 这里必须接收**kwargs，否则LangChain 0.3会注入config参数导致报错
        try:
            resp = requests.get(
                f"https://api.weatherapi.com/v1/current.json",
                params={"key": "YOUR_KEY", "q": city, "lang": "zh"},
                timeout=2.0
            )
            data = resp.json()
            return f"{city}当前温度: {data['current']['temp_c']}°C, 湿度: {data['current']['humidity']}%"
        except Exception as e:
            return f"天气查询失败: {str(e)}"

# 注册表
tool_registry = {
    "weather": WeatherTool(),
    "math": MathTool(),  # 另一个工具，省略实现
}

from collections import OrderedDict
import time, json

class DecayMemory:
    def __init__(self, max_entries=20, decay_hours=24):
        self.cache = OrderedDict()  # {key: (value, timestamp)}
        self.max_entries = max_entries
        self.decay_hours = decay_hours

    def add(self, key, value):
        # 插入时记录时间戳
        now = time.time()
        if key in self.cache:
            del self.cache[key]
        self.cache[key] = (value, now)
        # LRU淘汰：超出容量最久未用
        if len(self.cache) > self.max_entries:
            self.cache.popitem(last=False)

    def get_recent(self, k=5):
        """返回最近k条，但按时间衰减加权"""
        now = time.time()
        result = []
        for key, (val, ts) in reversed(self.cache.items()):
            age_hours = (now - ts) / 3600.0
            weight = max(0.0, 1.0 - age_hours / self.decay_hours)
            if weight > 0.3:  # 低于0.3权重直接丢弃
                result.append((key, val))
            if len(result) >= k:
                break
        return result

import random, asyncio

async def run_with_retry(func, *args, max_retries=3, base_delay=0.5):
    for attempt in range(max_retries):
        try:
            result = await func(*args)
            return result
        except Exception as e:
            if attempt == max_retries - 1:
                return f"[FATAL] 工具调用失败超过{max_retries}次: {str(e)}"
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1)
            print(f"[RETRY] 第{attempt+1}次失败，{delay:.2f}s后重试...")
            await asyncio.sleep(delay)

class CustomAgentExecutor(AgentExecutor):
    def _should_continue(self, steps, iterations):
        # 1. 最大迭代次数（硬性）
        if iterations >= self.max_iterations:  # 设置为8
            return False
        # 2. 重复工具检测：连续3次调用同一工具且参数相同
        if len(steps) >= 3:
            recent = [s.tool for s in steps[-3:]]
            if len(set(recent)) == 1:
                print("[STOP] 检测到同一工具连续调用3次，终止循环")
                return False
        # 3. 结果无效检测：连续2次返回相同结果
        if len(steps) >= 2:
            last_two = [s.observation for s in steps[-2:]]
            if last_two[0] == last_two[1]:
                print("[STOP] 连续两次结果相同，认为陷入死循环")
                return False
        return True