从 20 行最小闭环到"工程化 Agent"：我怎么一步步把它做成可演进的 Runtime

先跑通，再抽象；先能工作，再谈工程化

很多人一聊 Agent，就会先聊规划、记忆、工具调用、工作流，概念一个比一个大。但真到自己动手的时候，反而容易卡住：到底该先做什么？

我的做法比较简单 - 先别急着设计一整套宏大架构，先用一个最小闭环把“模型决策 - 系统执行 - 结果回流”这条链路跑通。最开始那版代码很短，甚至有点粗暴，但它把最关键的事情验证出来了：模型不只是会回答问题，它开始能真的参与执行任务。

后面所谓的工程化，并不是一开始凭空设计出来的，而是协议不够用了、状态开始混了、任务变复杂了，系统才一点点长出了 Planning、Tool Use、Memory 和 Runtime 这些结构。下面这篇文章，我就结合代码，把这条迭代路线完整复盘一遍。

---

1. 先别谈架构，先把最小闭环跑起来

我最开始验证的，不是什么复杂系统，而是一条最基本的循环：

用户提需求
-> 模型决定下一步
-> 系统执行
-> 把结果喂回模型
-> 直到任务完成

为了先看清逻辑，我连真实模型都没接，先写了一个假的 fake_llm()。

import os

messages = [
    {
        "role": "system",
        "content": "你必须只用两种格式回复：命令：xxx 或 完成：xxx"
    }
]


def fake_llm(messages):
    last = messages[-1]["content"]

    if "创建 hello.txt" in last:
        return '命令：echo "hello" > hello.txt'

    if "执行结果：" in last:
        return "完成：hello.txt 已经创建完成"

    return "完成：暂时不知道该怎么处理"


while True:
    user_input = input("你: ").strip()
    if user_input.lower() in {"exit", "quit"}:
        break

    messages.append({"role": "user", "content": user_input})

    while True:
        reply = fake_llm(messages)
        print("agent:", reply)

        if reply.startswith("命令："):
            command = reply.split("命令：", 1)[1].strip()
            result = os.popen(command).read()
            messages.append({"role": "user", "content": f"执行结果：{result}"})
        elif reply.startswith("完成："):
            break

这段代码虽然很简陋，但它第一次把 Agent 的最小逻辑闭环跑出来了。

以前是“你问一句，模型答一句”；现在变成了“你提一个任务，模型先判断要不要调用系统能力”。比如你输入 创建 hello.txt，它不只是回一句“你可以用 echo 命令”，而是会真的走“生成命令 -> 系统执行 -> 返回结果”这条链路。

2. 把 fake_llm 换成真实模型

逻辑跑通之后，下一步自然就是把 fake_llm() 换成真实模型。我这里用 OpenRouter 举例，核心代码也不复杂：

import os
import requests

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
MODEL_NAME = os.getenv("OPENROUTER_MODEL", "openai/gpt-4o-mini")


def call_openrouter(messages):
    url = "https://openrouter.ai/api/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {OPENROUTER_API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "temperature": 0.2,
    }

    resp = requests.post(url, headers=headers, json=payload, timeout=60)
    resp.raise_for_status()
    data = resp.json()
    return data["choices"][0]["message"]["content"]

然后把最小闭环里的 fake_llm(messages) 替换成 call_openrouter(messages) 就行。这时候，一个最原始的 Agent 已经有了雏形。

3. 第一个问题：字符串协议太脆了

最开始用“命令：xxx / 完成：xxx”这种协议，优点是直观，缺点也一样明显：它只适合演示，不适合往后长。因为你很快就会想加更多信息：这一步为什么这么做、调的是哪个工具、参数是什么、执行失败了怎么办、什么时候该结束。

所以我做的第一步升级，是把“命令字符串”改成“结构化动作”。

模型不再返回：

命令：echo "hello" > hello.txt

而是返回：

{
  "thought": "需要先创建文件",
  "action": "use_tool",
  "tool_name": "write_file",
  "tool_args": {
    "path": "hello.txt",
    "content": "hello"
  },
  "final_answer": null
}

对应的解析逻辑会变成这样：

import json

reply = """
{
  "thought": "需要先创建文件",
  "action": "use_tool",
  "tool_name": "write_file",
  "tool_args": {
    "path": "hello.txt",
    "content": "hello"
  },
  "final_answer": null
}
"""

decision = json.loads(reply)

if decision["action"] == "use_tool":
    print("调用工具:", decision["tool_name"])
    print("参数:", decision["tool_args"])
elif decision["action"] == "finish":
    print("最终答案:", decision["final_answer"])

这个改动看起来只是把文本换成 JSON，但本质上已经变了：以前系统是在“猜模型想干什么”；现在系统是在“读取一个明确的动作对象”。一旦动作结构化了，后面的日志、重试、调试、工具扩展，都会顺很多。

4. 第二个问题：别让模型直接碰系统，先收进工具箱

最开始直接 os.popen() 去执行模型给出的命令，演示起来很爽，但边界太松。所以后面我做的第二步升级，是把“系统能力”收拢成一组工具，再让模型从工具里选。

import os


def read_file(path: str) -> str:
    with open(path, "r", encoding="utf-8") as f:
        return f.read()


def write_file(path: str, content: str) -> str:
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"文件已写入：{path}"


def list_dir(path: str = "."):
    return sorted(os.listdir(path))


class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def register(self, name, func, description=""):
        self.tools[name] = {
            "func": func,
            "description": description,
        }

    def call(self, name, **kwargs):
        if name not in self.tools:
            raise ValueError(f"工具不存在: {name}")
        return self.tools[name]["func"](**kwargs)

    def specs(self):
        return [
            {"name": name, "description": meta["description"]}
            for name, meta in self.tools.items()
        ]


registry = ToolRegistry()
registry.register("read_file", read_file, "读取文本文件")
registry.register("write_file", write_file, "写入文本文件")
registry.register("list_dir", list_dir, "列出目录内容")

到这一步，系统才真正开始有“Tool Use”的味道：不是让模型直接碰底层系统，而是先定义清楚“系统能做什么”，再让模型在这些能力里做选择。

5. 第三个问题：messages 能记住，但记得太乱

最开始所有状态都放在 messages 里，这样当然可以跑，但任务一复杂，上下文就会越来越糊。因为不同信息的作用其实完全不一样：有的是最近对话，有的是当前任务计划，有的是工具执行结果，有的是以后可能复用的信息。

所以后面我开始把状态分层：

class Memory:
    def __init__(self, history_limit=10):
        self.short_term = []
        self.working = {
            "plan": None,
            "tool_results": [],
            "notes": []
        }
        self.long_term = {}
        self.history_limit = history_limit

    def add_message(self, role, content):
        self.short_term.append({"role": role, "content": content})
        self.short_term = self.short_term[-self.history_limit:]

    def set_plan(self, plan):
        self.working["plan"] = plan

    def add_tool_result(self, result):
        self.working["tool_results"].append(result)

    def add_note(self, note):
        self.working["notes"].append(note)

    def build_context(self):
        return {
            "recent_history": self.short_term,
            "working_memory": self.working,
            "long_term_memory": self.long_term,
        }

拆完之后，状态组织就清楚很多：short_term 管最近几轮对话，working 管当前任务做到哪一步，long_term 预留给以后更长期的记忆。到这一步，messages 就不再是唯一状态源了，而只是短期记忆的一部分。

6. 第四个问题：复杂任务不能只靠“现想现做”

最小闭环最适合的是“创建一个文件”“读取一个文件”“列一下目录”这种简单任务。但只要任务变成“先找到某个文件，再提取内容，再整理成摘要，再写到新文件里”，就会发现系统缺了一个很重要的东西：规划。

所以后面我加了一个最简单的 Planner。它做两件事：先生成高层计划；然后每一步根据当前状态决定下一步动作。

def create_plan(user_input: str):
    # 真实场景下可以让模型来生成
    return {
        "goal": user_input,
        "steps": [
            "理解用户任务",
            "判断是否需要调用工具",
            "执行工具并记录结果",
            "整理最终答案"
        ]
    }


plan = create_plan("帮我创建 hello.txt，并确认内容是否写入成功")
print(plan)

有了 plan 之后，系统就不再是“想到哪做到哪”，而是开始有了一个路线。

7. 把这些东西串起来：一个简化版 Agent Runtime

前面这些模块拆开看都不复杂，真正关键的是把它们串起来。为了便于展示，我这里给一个压缩在单文件里的“工程化简版”。真实项目里更合理的做法当然是拆成多个模块，但博客里先看清主流程更重要。

import os
import json
import re
import requests

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
MODEL_NAME = os.getenv("OPENROUTER_MODEL", "anthropic/claude-sonnet-4.6")


def extract_json(text: str):
    text = text.strip()
    match = re.search(r"```json\s*(\{.*\})\s*```", text, flags=re.S)
    if match:
        text = match.group(1)
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        match = re.search(r"\{.*\}", text, flags=re.S)
        if not match:
            raise ValueError(f"模型未返回合法 JSON：{text}")
        return json.loads(match.group(0))


class ToolRegistry:
    def __init__(self):
        self.tools = {}

    def register(self, name, func, description=""):
        self.tools[name] = {"func": func, "description": description}

    def call(self, name, **kwargs):
        if name not in self.tools:
            raise ValueError(f"工具不存在: {name}")
        return self.tools[name]["func"](**kwargs)

    def specs(self):
        return [
            {"name": name, "description": meta["description"]}
            for name, meta in self.tools.items()
        ]


class Memory:
    def __init__(self, history_limit=10):
        self.short_term = []
        self.working = {"plan": None, "tool_results": []}
        self.long_term = {}
        self.history_limit = history_limit

    def add_message(self, role, content):
        self.short_term.append({"role": role, "content": content})
        self.short_term = self.short_term[-self.history_limit:]

    def set_plan(self, plan):
        self.working["plan"] = plan

    def add_tool_result(self, result):
        self.working["tool_results"].append(result)

    def build_context(self):
        return {
            "recent_history": self.short_term,
            "working_memory": self.working,
            "long_term_memory": self.long_term,
        }

完整实现里，还会继续接上 Planner、LLM 调用和 AgentRuntime 的调度循环。但就算只看到这一步，也已经能明显看出几个模块的分工了：Planner 负责“先想清楚”，ToolRegistry 负责“系统有哪些能力”，Memory 负责“当前状态记到哪”，Runtime 负责“把整个循环串起来”。

8. 如果继续往工程里长，目录可以怎么拆

到这一步，代码已经不太适合一直塞在一个脚本里了。一个更自然的拆法是：

agent_runtime/
├── main.py
├── server.py
├── requirements.txt
└── agent/
    ├── __init__.py
    ├── llm.py
    ├── memory.py
    ├── metrics.py
    ├── planner.py
    ├── prompts.py
    ├── runtime.py
    ├── schema.py
    └── tools.py

这样一来，文章里那条迭代线也就很清楚了：最开始是一个“命令循环脚本”，后面慢慢长成了一个有规划、有工具、有记忆、有执行循环的 Agent Runtime。

9. 运行前要准备什么

安装依赖：

pip install requests

Linux / macOS 设置环境变量：

export OPENROUTER_API_KEY="你的api-key"
export OPENROUTER_MODEL="anthropic/claude-sonnet-4.6"
python agent_demo.py

Windows PowerShell：

$env:OPENROUTER_API_KEY="你的api-key"
$env:OPENROUTER_MODEL="anthropic/claude-sonnet-4.6"
python agent_demo.py

10. 结尾

一开始，我只是想让模型别光聊天，能真的替我干点事。后来我发现，真要让它稳定干活，就不能只靠一轮回答和一条命令。再后来，每一个“不够用”的地方，都逼着系统长出了新的结构。最后它就从一个最小逻辑闭环，慢慢变成了一个有规划、有工具、有状态管理的 Agent Runtime。

很多时候，工程化不是一开始就设计得多完整，而是先把最小链路跑通，再让真实问题逼着结构长出来。