LangChain 0.3手写Agent：工具注册表与记忆衰减的12个工程细节

python=3.11.5
langchain=0.3.7
langchain-openai=0.2.8
pydantic=2.7.4
redis=5.0.7  # 用于外部记忆存储

# tools.py
from langchain_core.tools import tool
from pydantic import BaseModel, Field
import numpy_financial as npf

class IRRInput(BaseModel):
    cash_flows: list[float] = Field(description="现金流序列，负数代表投资")
    periods: int = Field(description="期数", ge=1, le=60)

@tool("calc_irr", args_schema=IRRInput)
def calc_irr(cash_flows: list[float], periods: int) -> str:
    """计算内部收益率，返回百分比格式"""
    try:
        rate = npf.irr(cash_flows)
        return f"{rate*100:.2f}%"
    except Exception:
        return "ERROR: 现金流无法收敛，请检查输入"

class QueryInput(BaseModel):
    sql: str = Field(description="SQL查询语句")
    db_name: str = Field(description="目标数据库")

@tool("query_db", args_schema=QueryInput)
def query_db(sql: str, db_name: str) -> str:
    """执行只读SQL查询，禁止写入操作"""
    # 实际接入PostgreSQL连接池
    result = execute_read_only(sql, db_name)  
    return str(result[:200])  # 截断防止token爆炸

tool_registry = {"calc_irr": calc_irr, "query_db": query_db}

# agent.py
from collections import deque
import redis, json, time

class MemoryTier:
    def __init__(self, redis_host="localhost", port=6379):
        self.short_mem = deque(maxlen=8)  # 滑动窗口
        self.r = redis.Redis(host=redis_host, port=port, db=0)

    def add_short(self, role: str, content: str):
        self.short_mem.append({"role": role, "content": content, "ts": time.time()})

    def add_long(self, key: str, value: dict, ttl_hours=24):
        # 长期记忆，带权重衰减
        value["weight"] = 1.0
        value["last_access"] = time.time()
        self.r.hset(f"agent:{key}", mapping=value)
        self.r.expire(f"agent:{key}", ttl_hours * 3600)

    def get_long(self, key: str) -> dict | None:
        data = self.r.hgetall(f"agent:{key}")
        if not data:
            return None
        # 遗忘曲线：超过12小时，权重减半
        elapsed = time.time() - float(data[b"last_access"])
        if elapsed > 12 * 3600:
            data[b"weight"] = float(data[b"weight"]) * 0.5
        return data

def agent_loop(task: str, max_iterations: int = 8):
    mem = MemoryTier()
    prompt = build_prompt()  # 包含工具描述和记忆格式化

    for i in range(max_iterations):
        # 组装当前对话上下文
        context = format_memory(mem.short_mem, mem.get_long("constraints"))
        response = llm.invoke(prompt.format(task=task, 
                                            context=context, 
                                            tools=list(tool_registry.keys())))

        # 三级错误恢复
        parsed = parse_action(response.content)
        if parsed is None:
            # 第一级：尝试修复JSON
            parsed = repair_json(response.content)
            if parsed is None:
                # 第二级：重试一次
                response = llm.invoke(prompt.format(task=task, 
                                                    context=context, 
                                                    tools=list(tool_registry.keys())))
                parsed = parse_action(response.content)
                if parsed is None:
                    # 第三级：保守回退
                    return {"result": "解析失败，建议简化任务", "iterations": i}

        if parsed["action"] == "finish":
            return {"result": parsed["args"]["output"], "iterations": i}

        # 执行工具并记录记忆
        try:
            tool_fn = tool_registry[parsed["action"]]
            result = tool_fn.invoke(parsed["args"])
            mem.add_short("assistant", f"调用{parsed['action']}: {parsed['args']}")
            mem.add_short("system", f"结果: {result}")
        except KeyError:
            mem.add_short("system", f"ERROR: 工具不存在，可用的有{list(tool_registry)}")
            continue