挑战项目 A · 手搓一个迷你 Agent 框架

   你会亲手实现这些"框架黑箱"里的东西：
   ① 工具注册与调度（装饰器自动生成 schema）
   ② 上下文压缩（对话满了自动总结重启）——第 4 节
   ③ 子 Agent 派生（隔离上下文，只回摘要）——第 4、13 节
   ④ 护栏（max_steps / 超时 / 预算上限）——第 1 节
   ⑤ 可观测性（结构化 trace，能回放每一步）——第 14 节

   做完你会得到一个"啊，原来 LangChain 就这？"的顿悟。
   之后用任何框架，你都知道皮下是什么，出问题知道去哪找。

   MiniAgent
   ├── @tool 装饰器 ── 自动从函数签名和 docstring 生成工具 schema
   ├── run() 主循环 ── 第1节的循环，加上：
   │     ├── 每轮检查 token 预算，超阈值 → compact()
   │     ├── max_steps / 超时护栏
   │     └── 结构化 trace 记录每一步
   ├── compact() ── 历史 → 摘要，保留关键事实
   ├── spawn_subagent() ── 开一个干净上下文的子 Agent，只回摘要
   └── Tracer ── 每步的 (思考/工具调用/结果) 存成可回放的 JSON

import inspect, json, time
from typing import Callable

# ============ @tool 装饰器：自动生成 schema ============
def tool(fn: Callable):
    sig = inspect.signature(fn)
    params = {
        name: {"type": "string", "desc": ""}  # 简化：都当 string，真实版按注解映射
        for name in sig.parameters
    }
    fn._schema = {
        "name": fn.__name__,
        "description": (fn.__doc__ or "").strip(),
        "parameters": params,
    }
    return fn

# ============ 迷你 Agent ============
class MiniAgent:
    def __init__(self, model, tools, max_steps=20, token_budget=50_000,
                 compact_at=0.75, timeout_s=300):
        self.model = model
        self.tools = {t._schema["name"]: t for t in tools}
        self.max_steps = max_steps
        self.token_budget = token_budget
        self.compact_at = compact_at
        self.timeout_s = timeout_s
        self.messages = []
        self.trace = []

    def _est_tokens(self):
        return sum(len(m["content"]) for m in self.messages) // 3  # 粗估

    def compact(self):
        """历史 → 摘要，保留关键事实（框架的灵魂）"""
        history = "\n".join(f"{m['role']}: {m['content']}" for m in self.messages)
        prompt = (f"总结以下 Agent 工作历史，必须保留：已完成的事、当前状态、"
                  f"下一步计划、所有关键事实（文件路径/决策/报错原文）。\n\n{history}")
        summary = self.model.chat([{"role": "user", "content": prompt}])
        # 只保留 system + 摘要，历史被压缩掉
        system = self.messages[0]
        self.messages = [system, {"role": "user",
                                  "content": f"【历史摘要】\n{summary}\n\n请继续。"}]
        self.trace.append({"event": "compact", "summary_len": len(summary)})

    def run(self, task, system_prompt):
        self.messages = [{"role": "system", "content": system_prompt},
                         {"role": "user", "content": task}]
        start = time.time()
        for step in range(self.max_steps):
            if time.time() - start > self.timeout_s:
                return self._finish("超时护栏触发", step)
            if self._est_tokens() > self.token_budget * self.compact_at:
                self.compact()

            reply = self.model.chat(self.messages)
            self.messages.append({"role": "assistant", "content": reply})

            call = self._parse_tool_call(reply)
            if call is None:
                self.trace.append({"step": step, "type": "final", "content": reply})
                return reply

            tool_fn = self.tools.get(call["tool"])
            result = (tool_fn(**call["args"]) if tool_fn
                      else f"错误：无 {call['tool']} 工具")
            self.messages.append({"role": "user", "content": f"工具结果：\n{result}"})
            self.trace.append({"step": step, "type": "tool",
                               "call": call, "result": result[:200]})
        return self._finish("达到最大步数", self.max_steps)

    def spawn_subagent(self, task, tools, system_prompt):
        """派生子 Agent：干净上下文，只回摘要"""
        sub = MiniAgent(self.model, tools, max_steps=self.max_steps)
        result = sub.run(task, system_prompt)
        self.trace.append({"event": "subagent", "task": task[:80],
                           "sub_steps": len(sub.trace)})
        return result  # 只有这个摘要回到主 Agent，子 Agent 的 8000 字历史被丢弃

    def _parse_tool_call(self, reply):
        try:
            c = json.loads(reply.strip())
            return c if "tool" in c else None
        except json.JSONDecodeError:
            return None

    def _finish(self, reason, step):
        self.trace.append({"event": "halt", "reason": reason, "step": step})
        return f"[{reason}] 已完成步骤：{step}"