Reborn 的技术博客

一个最简化的 AI Agent 核心实现

2026-06-23·AI, Agent, Python, LLM

依赖：openai / 模块：AgentTrace（可观测）、Compaction（上下文压缩）、Persistence（持久化）

一、整体架构

agent.py (~300 行)
  ├── trace.py      ── 可观测性：Span 树 + Trace 日志
  ├── compaction.py ── 上下文压缩：超长时自动摘要
  ├── persistence.py ── 持久化：会话保存/恢复 (Store → jsonl)
  └── openai          ── LLM 调用 (兼容接口)

一条线串起 4 个层次 —— 最简 Agent → 可观测 → 上下文管理 → 持久化 —— 每一层只引入一个类/文件，不搞抽象。

二、工具系统

2.1 工具即函数

def search_web(query: str) -> str:
    """模拟搜索，实际可接 Google/Bing API"""
    fake_db = {"北京": "北京今天晴天，25°C~32°C", ...}
    return fake_db.get(query, f"未找到'{query}'的相关结果")

def calculate(expression: str) -> str:
    """安全的数学计算"""
    ...

2.2 JSON Schema 注册表

遵循 OpenAI Function Calling 规范：

TOOLS = [
    {"type": "function", "function": {"name": "search_web", "description": "...", "parameters": {...}}},
    {"type": "function", "function": {"name": "calculate",   "description": "...", "parameters": {...}}},
]

TOOL_MAP = {"search_web": search_web, "calculate": calculate}

设计要点：LLM 只读 TOOLS 的 JSON Schema 做推理，运行时由 TOOL_MAP 分发执行——模型无法直接跑代码，安全隔离。

三、核心循环：run_agent()

def run_agent(user_input: str, max_steps: int = 5) -> str:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_input},
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model="your-model-name",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",   # LLM 自己决定要不要调工具
        )
        msg = response.choices[0].message

        if not msg.tool_calls:           # 情况 A：直接回答
            return msg.content

        messages.append(msg)             # 情况 B：执行工具

        for tool_call in msg.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            result = TOOL_MAP[name](**args)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })
        # 循环回去，LLM 看到工具结果后继续思考

    return "⚠️ 达到最大步数限制"

设计要点

| 要点 | 说明 | | --- | --- | | tool_choice="auto" | LLM 自主决策，不强制调用 | | 对话历史 = 状态 | messages 数组是 Agent 的全部「记忆」 | | 最大步数 5 | 硬上限防无限循环 | | 纯文本交互 | 无 UI 依赖，stdio 打印每步 |

四、增强版：run_agent_with_trace()

在基础循环上叠加三层能力：

  1. 可观测 —— AgentTracer：log_llm_call + log_tool_call 记录延迟/输入/输出，构建 Span 树
  2. 上下文压缩 —— ContextManager：ctx.maybe_compact(messages) 超长时自动摘要
  3. 持久化 —— PersistenceManager：new_session_id + save_session + load_session + 每轮 save_messages 防崩溃
def run_agent_with_trace(user_input, *, tracer, client, ctx, pm, session_id=None, max_steps=5):
    # ── 创建或恢复会话 ──
    if session_id:
        state = pm.load_session(session_id)
        messages = state["messages"]
        ctx.restore(state["summary"])
    else:
        session_id = pm.new_session_id()
        messages = []

    messages.append({"role": "user", "content": user_input})
    # ... 主循环，每轮记录 llm_span / tool_span，检查压缩，自动保存

五、整体设计评价

优点：

  1. 极简：核心 Agent 循环 ~50 行
  2. 渐进增强：基础 → trace → compaction → persistence，每层独立模块化
  3. 安全：工具执行在 host 端，LLM 只输出 tool_call JSON
  4. 可恢复：full session 持久化 + 单轮自动保存
  5. OpenAI 兼容：base_url 可替换为任何兼容 API

扩展方向：

  • Skill 渐进式加载（按需激活工具）
  • 多轮对话循环（会话管理）
  • 流式输出
  • 容错机制（重试 + 超时）
#AI#Agent#Python#LLM