从零手写AI Agent：LangChain工具链与循环控制实现细节

from langchain_core.tools import tool
from pydantic import BaseModel, Field
import json, math

class CalculatorInput(BaseModel):
    expression: str = Field(description="一个合法的Python数学表达式，例如 '(3+5)*2'")

@tool("calculator", args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """计算数学表达式。仅支持 + - * / ** sqrt 和括号，不支持变量赋值。"""
    allowed = {"sqrt": math.sqrt, "__builtins__": {}}
    try:
        result = eval(expression, allowed, {})
        return f"计算结果: {result}"
    except Exception as e:
        return f"Error: 表达式无法计算，原因: {type(e).__name__}: {e}"

@tool("search_kb")
def search_kb(query: str) -> str:
    """在本地知识库中检索。输入自然语言问题，返回最相关的3条文本片段。"""
    # 这里用假数据模拟，实际接向量库
    kb = {"退款": "退款需在购买后7天内申请，超过7天不支持。",
          "发票": "发票在订单完成后3个工作日内开具，可在账户中心下载。"}
    hits = [v for k, v in kb.items() if k in query]
    return "\n".join(hits) if hits else "未检索到相关内容。"

from langchain_core.messages import BaseMessage, SystemMessage

class WindowMemory:
    def __init__(self, max_tokens: int = 3000, keep_last: int = 12):
        self.history: list[BaseMessage] = []
        self.max_tokens = max_tokens
        self.keep_last = keep_last

    def add(self, msg: BaseMessage):
        self.history.append(msg)

    def get(self) -> list[BaseMessage]:
        # 粗粒度token估算：字符数/3.5，中文场景偏保守
        while len(self.history) > self.keep_last:
            est = sum(len(str(m.content)) for m in self.history) / 3.5
            if est  str:
        self.memory.add(SystemMessage(content=(
            "你是一个严谨的助手。优先使用工具获取事实，不要编造。"
            "如果工具返回Error，请先分析原因，最多重试2次，仍失败则向用户说明。")))
        self.memory.add(HumanMessage(content=user_input))

        for step in range(self.max_steps):
            resp = self.llm_with_tools.invoke(self.memory.get())
            self.memory.add(resp)

            # 终止条件：模型没有调用工具，直接给出最终答案
            if not resp.tool_calls:
                return resp.content

            for call in resp.tool_calls:
                name, args = call["name"], call["args"]
                tool = self.tools.get(name)
                if tool is None:
                    result = f"Error: 工具 '{name}' 不存在，可用工具: {list(self.tools)}"
                else:
                    try:
                        result = tool.invoke(args)
                    except Exception as e:
                        result = f"Error: 工具执行异常 {type(e).__name__}: {e}"
                self.memory.add(ToolMessage(content=str(result), tool_call_id=call["id"]))

        return "已达到最大推理步数，任务未能完成。请缩小问题范围后重试。"