基于LangChain 0.3手写AI Agent：工具定义、记忆管理与循环控制实现

python==3.11.9
langchain==0.3.7
langchain-core==0.3.15
langchain-openai==0.2.8
openai==1.54.0
pydantic==2.9.2

用户输入
  ↓
[记忆组装] → 拼接 system prompt + 历史窗口 + 工具描述
  ↓
[LLM决策] → 输出 Thought + Action(工具名+参数) 或 Final Answer
  ↓
[工具执行] → 参数校验 → 执行 → 异常捕获 → 结果截断
  ↓
[循环控制] → 判断终止条件（Final / 超步数 / 超时 / 超预算）
  ↓
回到[记忆组装]，把 Observation 追加进scratchpad

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

class CalcInput(BaseModel):
    expression: str = Field(description="要计算的数学表达式，例如 '(1200*0.85)+350'，仅支持四则运算和括号")

@tool("calculator", args_schema=CalcInput)
def calculator(expression: str) -> str:
    """当你需要进行数值计算时使用此工具。不要自己心算，所有算术一律走这个工具。"""
    allowed = set("0123456789+-*/(). ")
    if not set(expression)  str:
    """查询指定城市当前天气。仅当用户明确询问天气、温度、是否下雨时使用。"""
    # 模拟外部API，故意留10%失败率用于测试错误处理
    if time.time() % 10  str:
        start = time.time()
        scratchpad = []
        last_call, repeat_count = None, 0

        for step in range(self.max_iterations):
            # 终止条件1：超时
            if time.time() - start > self.max_seconds:
                return f"[超时终止] 已执行{step}步，最后观察：{scratchpad[-1] if scratchpad else '无'}"
            # 终止条件2：token预算
            if self.total_tokens > self.token_budget:
                return f"[预算终止] 累计{self.total_tokens}tokens"

            msgs = self._build_messages(user_input, scratchpad)
            step_out: AgentStep = structured_llm.invoke(msgs)
            self.total_tokens += sum(len(m.content) for m in msgs) // 3  # 粗略估算

            if step_out.action == "final_answer":
                answer = step_out.action_input.get("answer", "")
                self.history.append((user_input, answer))
                return answer

            # 终止条件3：死循环检测
            call_sig = (step_out.action, json.dumps(step_out.action_input, sort_keys=True))
            if call_sig == last_call:
                repeat_count += 1
                if repeat_count >= 2:
                    return f"[循环终止] 工具{step_out.action}连续重复调用"
            else:
                repeat_count, last_call = 0, call_sig

            observation = self._execute_tool(step_out.action, step_out.action_input)
            scratchpad.append({
                "thought": step_out.thought,
                "action": step_out.action,
                "action_input": step_out.action_input,
                "observation": observation,
            })

        return f"[步数终止] 达到max_iterations={self.max_iterations}"

    def _execute_tool(self, name: str, args: dict) -> str:
        tool = TOOL_MAP.get(name)
        if tool is None:
            return f"ERROR: 未知工具 {name}，可用工具：{list(TOOL_MAP)}"
        for attempt in range(2):  # tool_retry=2
            try:
                result = tool.invoke(args)
                return str(result)[:2000]  # 截断，防上下文爆炸
            except Exception as e:
                if attempt == 1:
                    return f"ERROR: {type(e).__name__}: {str(e)[:200]}"
                time.sleep(0.5)
        return "ERROR: unreachable"

agent = SimpleAgent()
print(agent.run("杭州今天天气怎么样？如果气温高于20度，帮我算一下 (22*1.8)+32 是多少"))
# 典型输出（2-3步）：
# 多云 22℃ → 22>20 → calculator('(22*1.8)+32') → 71.6