LangChain 0.3手写Agent：工具注册与记忆衰减机制全解

Python 3.11.7
langchain==0.3.7
langchain-openai==0.2.14
openai==1.52.0
faiss-cpu==1.8.0.post1

┌─────────────────────────────────────┐
│  Control Layer: 循环计数器/熔断器    │
├─────────────────────────────────────┤
│  Memory Layer: 带TTL的向量存储      │
├─────────────────────────────────────┤
│  Tool Layer: 装饰器注册+参数校验    │
└─────────────────────────────────────┘

# tools.py
from functools import wraps
from typing import Callable, Dict, Any, List
from pydantic import BaseModel, ValidationError
import inspect

class ToolRegistry:
    """工具注册中心：通过装饰器注册，自动生成schema"""

    def __init__(self):
        self.tools: Dict[str, Dict[str, Any]] = {}

    def register(self, name: str, description: str, 
                 parameters_schema: BaseModel):
        """装饰器工厂"""
        def decorator(func: Callable):
            @wraps(func)
            def wrapper(**kwargs):
                # 参数校验：防止LLM幻觉参数
                try:
                    validated = parameters_schema(**kwargs)
                except ValidationError as e:
                    return f"参数错误: {e.errors()}"
                # 实际调用
                return func(**validated.model_dump())

            # 注册到工具表
            self.tools[name] = {
                "name": name,
                "description": description,
                "parameters": parameters_schema.schema(),  # JSON Schema
                "func": wrapper
            }
            return wrapper
        return decorator

# 示例：查询合同工具
from pydantic import BaseModel, Field

class ContractQueryParams(BaseModel):
    contract_id: str = Field(..., description="合同编号，格式如HT-2024-001")
    query_type: str = Field(..., description="查询类型: amount/status/parties")

registry = ToolRegistry()

@registry.register(
    name="query_contract",
    description="查询供应链金融合同的关键信息，如金额、状态、签约方",
    parameters_schema=ContractQueryParams
)
def query_contract(contract_id: str, query_type: str) -> str:
    """模拟数据库查询，实际项目替换为DB调用"""
    # 模拟不同合同数据
    mock_db = {
        "HT-2024-001": {"amount": 500000, "status": "已放款", "parties": "A公司/B公司"},
        "HT-2024-002": {"amount": 1200000, "status": "审核中", "parties": "C公司/D公司"}
    }
    contract = mock_db.get(contract_id)
    if not contract:
        return f"未找到合同 {contract_id}"
    return f"合同{contract_id}的{query_type}是: {contract.get(query_type, '未知')}"

# memory.py
import time
import numpy as np
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

class TTLVectorMemory:
    """带生存时间(TTL)的向量记忆，防止上下文爆炸"""

    def __init__(self, ttl_seconds: int = 1800, max_entries: int = 100):
        self.ttl_seconds = ttl_seconds
        self.max_entries = max_entries
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-small",
            dimensions=512  # 降维，减少token消耗
        )
        self.vs = None
        self._entries = []  # [(timestamp, text)]

    def add(self, text: str):
        """添加记忆，自动清理过期条目"""
        now = time.time()
        # 过期清理
        self._entries = [(t, txt) for t, txt in self._entries 
                        if now - t = self.max_entries:
            self._entries.pop(0)
            # 重建向量库（简化，实际可增量）
            self.vs = None

        self._entries.append((now, text))

        # 向量化存储
        if self.vs is None:
            self.vs = FAISS.from_texts(
                [text], self.embeddings
            )
        else:
            self.vs.add_texts([text])

    def search(self, query: str, top_k: int = 3) -> List[str]:
        """相似度检索，只取最相关的记忆"""
        if self.vs is None:
            return []
        results = self.vs.similarity_search(query, k=top_k)
        return [doc.page_content for doc in results]

# agent.py
from langchain_openai import ChatOpenAI
from langchain.schema import SystemMessage, HumanMessage

class MiniAgent:
    """300行代码的最小可运行Agent"""

    def __init__(self, registry: ToolRegistry, 
                 max_steps: int = 5, 
                 max_same_tool_calls: int = 2):
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0.2,
            max_tokens=1024
        )
        self.registry = registry
        self.max_steps = max_steps
        self.max_same_tool_calls = max_same_tool_calls
        self.memory = TTLVectorMemory()

        # 工具调用历史，用于熔断
        self.tool_history = []

    def run(self, user_query: str) -> str:
        """主循环：计划-执行-反思"""
        messages = [
            SystemMessage(content=self._build_system_prompt()),
            HumanMessage(content=user_query)
        ]

        step = 0
        while step  bool:
        """熔断器：连续N次调用相同工具则阻断"""
        if len(self.tool_history) < self.max_same_tool_calls:
            return True
        recent = self.tool_history[-self.max_same_tool_calls:]
        return not all(t == tool_name for t in recent)