从零手写AI Agent：LangChain与AutoGPT的循环控制与记忆管理

while step  str:
    fake = {"北京": 18, "上海": 24, "深圳": 29}
    return f"{city}当前气温{fake.get(city, 20)}摄氏度"

register(Tool(
    name="get_weather",
    description="查询指定城市当前气温，参数city为中文城市名",
    func=get_weather,
    args_schema={"city": "str"}
))

def calc(expression: str) -> str:
    # 只允许数字和运算符，防止注入
    if not all(c in "0123456789+-*/.() " for c in expression):
        raise ValueError("非法字符")
    return str(eval(expression, {"__builtins__": {}}, {}))

register(Tool(
    name="calc",
    description="计算数学表达式，例如 '24-18'",
    func=calc,
    args_schema={"expression": "str"}
))

class Memory:
    def __init__(self, window=5, llm=None):
        self.window = window
        self.llm = llm
        self.buffer = []      # [(action_str, observation_str)]
        self.summary = ""

    def add(self, action, observation):
        self.buffer.append((action, observation))
        if len(self.buffer) > self.window:
            self._compress()

    def _compress(self):
        old = self.buffer[:-self.window // 2]
        self.buffer = self.buffer[-self.window // 2:]
        text = "\n".join(f"动作:{a} 结果:{o}" for a, o in old)
        prompt = f"用一句话总结以下Agent执行历史，保留关键数字和结论：\n{text}"
        self.summary = self.llm.invoke(prompt).content.strip()

    def render(self):
        lines = []
        if self.summary:
            lines.append(f"[历史摘要] {self.summary}")
        for a, o in self.buffer:
            lines.append(f"[动作] {a}\n[观察] {o}")
        return "\n".join(lines)

import time, re

def llm_with_retry(llm, prompt, max_retries=3, base=0.5):
    for i in range(max_retries):
        try:
            return llm.invoke(prompt)
        except Exception as e:
            if i == max_retries - 1:
                raise
            time.sleep(base * (2 ** i))
    raise RuntimeError("unreachable")

ACTION_RE = re.compile(r"Action:\s*(\w+)\s*Args:\s*(\{.*?\})", re.S)

def run_agent(task: str, llm, max_steps=6):
    memory = Memory(window=5, llm=llm)
    last_action = None
    repeat_count = 0

    for step in range(max_steps):
        prompt = build_prompt(task, memory, TOOLS)
        raw = llm_with_retry(llm, prompt).content
        print(f"[step {step}] raw={raw[:120]}")

        if "Final Answer:" in raw:
            return raw.split("Final Answer:")[-1].strip()

        m = ACTION_RE.search(raw)
        if not m:
            memory.add("(解析失败)", "输出格式错误，请使用 Action/Args 或 Final Answer")
            continue

        tool_name, args_raw = m.group(1), m.group(2)
        # 重复动作熔断
        sig = f"{tool_name}:{args_raw}"
        if sig == last_action:
            repeat_count += 1
            if repeat_count >= 2:
                return f"[熔断] 连续重复动作 {sig}，终止"
        else:
            repeat_count = 0
        last_action = sig

        if tool_name not in TOOLS:
            memory.add(sig, f"ERROR: 工具 {tool_name} 不存在")
            continue

        try:
            args = json.loads(args_raw)
            obs = TOOLS[tool_name].run(**args)
        except Exception as e:
            obs = f"ERROR: {type(e).__name__}: {e}"
        memory.add(sig, str(obs))

    return "[熔断] 达到最大步数"

可用工具：
- get_weather: 查询指定城市当前气温，参数city为中文城市名
- calc: 计算数学表达式，例如 '24-18'

请严格按格式输出：
Action: 工具名
Args: {"参数名": "值"}
或
Final Answer: 最终答案