Command Palette

Search for a command to run...

0

构建 Baby Harness:自进化的 AI Agent 任务框架

为什么写一个轻量级自治 Agent 框架,以及它的架构设计全貌——零外部依赖,1500 行 Python,16 个模块,148 个测试。

构建 Baby Harness:自进化的 AI Agent 任务框架

为什么写一个 AI Agent 框架,以及它的架构设计全貌


缘起

一切从一个很简单的想法开始:

"设个目标,它自己干到完,越干越熟。"

我日常的工作流是用 Hermes Agent(一个 AI 助手)来开发项目。每次我都需要告诉它做什么、怎么做、改完了跑测试、失败了重来——这是一个重复的循环。

为什么不把这个循环自动化?

我需要一个系统:

  1. 给它一个目标,它能自动分解为子任务
  2. 逐个执行,失败自动重试
  3. 记录每次的教训,下次做得更好
  4. 跑得越久,结果越好

现有的选择也不少:LangChain、AutoGPT、CrewAI、Smolagents。但它们太重了。一个简单的自动化循环,不需要向量数据库、不需要复杂的 Agent Graph、不需要编排引擎。

于是我写了 Baby Harness


设计原则

从一开始就定了几个铁律:

原则原因
零外部依赖(除了 Pydantic)轻量部署,pip install 就能用
Generator-Verifier 分离像代码审查一样,写代码的和检查代码的分开
自进化(不搞向量库)SharedMemory + 简单相似度就够了
TDD 驱动每个功能先写测试,后写代码
小文件、小模块每个模块<100行,容易理解和修改

最终结果:16 个模块,~1500 行 Python,零外部依赖。


架构全景

                    ┌─────────────┐
                    │  CLI / API  │
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │ Coordinator │  (重试逻辑 + 自进化触发)
                    └──────┬──────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
   ┌──────▼──────┐  ┌─────▼──────┐  ┌──────▼──────┐
   │  TaskQueue  │  │ TeamRegistry│  │SharedMemory │
   │ (JSON文件)  │  │            │  │ (JSON文件)  │
   └─────────────┘  └─────┬──────┘  └─────────────┘
                          │
                   ┌──────▼──────┐
                   │  AgentTeam  │
                   │ ┌────────┐  │
                   │ │Memory  │  │  (经验池、技能、
                   │ │(进化)  │  │   成功率追踪)
                   │ └────────┘  │
                   │              │
              ┌────▼────┐   ┌────▼────┐
              │Executor │   │Validator│
              │(通用)   │   │(通用)   │
              └─────────┘   └─────────┘

核心模块

1. Models (models.py) — 数据模型

所有模块的数据契约。Task、ExecutionOutput、ValidationResult、Learning。

class Task(BaseModel):
    id: str
    goal: str                    # 任务目标
    status: TaskStatus           # pending → running → done/failed
    tags: list[str]              # domain 路由用
    max_retries: int = 3
    timeout: int = 60

设计决策:没有 domaininstruction 字段。domain 从 tags[0] 解析。少一个字段就少一种复杂。

2. TaskQueue (queue.py) — 持久化任务队列

按 priority 排序的 FIFO 队列,next() 返回最高优先级待处理任务,自动标记 RUNNING。

queue.add(task)
task = queue.next()  # 取最高优先级的 pending 任务
queue.save()         # 持久化到 JSON

JSON 持久化是为了跨进程恢复——CLI 每次调用是新进程,没有文件就会丢数据。

3. AgentTeam (agent_team.py) + TeamRegistry

每个 AgentTeam 有自己的 domain、generator、validator、memory。TeamRegistry 按 domain 查找最佳团队。

team = AgentTeam(
    team_id="backend",
    domain="general",
    generator=MockExecutor(),
    validator=CompositeValidator([...]),
    memory=AgentMemory(agent_id="backend"),
)
registry.register(team)

进化机制

def evolve(self, task_result: dict) -> None:
    self.tasks_completed += 1
    self.success_rate = self.memory.get_success_rate()
    if task_result.get("success"):
        self.memory.add_experience(task_result.get("output", ""))

每次执行后更新成功率,累积经验。越用越准。

4. Executor (executor.py) — 3 种实现

Executor用途成本
MockExecutor测试,返回固定输出0
HermesExecutorhermes chat -q 执行中等 (spawn Hermes)
LLMExecutor直接调 DeepSeek/OpenAI API低 (直接 HTTP)

抽象接口只有一句话:

class Executor(ABC):
    @abstractmethod
    def execute(self, task: Task, context: str = "") -> ExecutionOutput: ...

5. SharedMemory (shared_memory.py) — "越干越熟"的基础

跨团队知识共享,tag 匹配 + 简单持久化。

sm.add_experience(
    experience="Task X failed because timeout < 3s",
    tags=["failure", "timeout"],
    team_id="backend",
    success=False,
)
 
# 后面遇到类似问题时:
related = sm.get_relevant_experiences(tags=["failure", "timeout"])

为什么不用向量库:在 solo developer 的场景,一年也攒不了几百条经验。顺序扫描 + 简单 tag 过滤绰绰有余。

6. Coordinator (coordinator.py) — 大脑

把上面所有东西串起来:

def execute_task(self, task: Task) -> ExecutionOutput:
    team = self.assign_task(task)
    context = ""
    for attempt in range(min(MAX_RETRIES, task.max_retries)):
        output = team.generator.execute(task, context=context)
        result = team.validator.validate(task, output)
        if result.passed:
            team.evolve({"success": True, "output": output})
            self.shared_memory.add_experience(...)
            return output
        context = result.message  # 注入验证反馈
    team.evolve({"success": False})
    return last_output

关键设计:重试时注入验证器的反馈作为 context。每次失败的尝试都是下一步的输入。

7. PromptCache (prompt_cache.py) — LLM 调用缓存

缓存 LLM 调用结果,key 设计参考了 LangChain 的 llm_string 模式:

# 之前(太简单):
key = model | prompt | context
 
# 之后(LangChain 模式):
key = sha256({model, messages[], temperature, max_tokens, ...})

这保证了 system prompt、temperature、max_tokens 全部进 key。不同配置不会互相污染缓存。

8. ContextManager (context_manager.py) — 长时间上下文控制

解决"跑一小时,context window 满了"的问题。

第10步: messages = [system] + [compact: 1-7步摘要] + [step 8,9,10详细] + [当前]
第20步: messages = [system] + [compact: 1-17步摘要] + [step 18,19,20详细] + [当前]
               ^^^^^^ 永远不涨过有效窗口(200K token)

灵感来自 Claude Code 的 /compact 和 LangChain 的 "Write, Select, Compress, Isolate" 四策略。


缓存系统的三层设计

LLM 调用的缓存优化是另一个重点,分了 3 层:

Layer 1: PromptCache(精确匹配)
  - LRU + TTL
  - key = sha256(model + messages + params)
  - 解决:retry 循环的重复调用

Layer 2: PatternReuse(语义匹配)
  - Overlap 系数查 SharedMemory
  - 同义 prompt 命中同一结果
  - 解决:"fix bug in X.ts" 和 "fix issue in X.ts" 应该一样

Layer 3: 自进化(越用越熟)
  - 成功的 prompt→response 存 SharedMemory
  - 下次类似任务先查记忆
  - SharedMemory 积累越多,命中越高

自进化的真实含义

"自进化"听起来很玄乎,但在 Baby Harness 里其实很简单:

第 1 次运行

目标: "给 cache.ts 加 LRU 淘汰"
  步骤: 读 cache.ts → 写代码 → 跑测试
  结果: ✅ 通过,prompt→response 存 SharedMemory

第 2 次运行(不同目标):

目标: "给 http.ts 加超时重试"
  步骤: 读 http.ts → 写代码 → 跑测试
  PromptCache: cache.ts 的 prompt 和 http.ts 不同 → miss
  PatternReuse: "加 LRU 淘汰"和"加超时重试"都含"加"、"ts" → 部分匹配
  SharedMemory: 有 cache.ts 的成功经验 → 参考写法
  结果: ✅ 更快完成

第 10 次运行

SharedMemory 积累了 50+ 条经验 → PatternReuse 命中率 > 30%

并不是 AI 在进化,而是系统在使用过程中积累了越来越精准的上下文记忆。


CLI & 自治代理

开发助手 (dev-harness.py)

python3 dev-harness.py test       # 跑测试 + 记录指标
python3 dev-harness.py status     # 看成功率趋势
python3 dev-harness.py learnings  # 看历史学习

自治开发代理 (goal.py)

python3 goal.py "添加Yandex搜索引擎"
python3 goal.py "优化中文搜索质量"
python3 goal.py "修掉所有类型错误"

自治代理的内部循环:

1. LLM 解析目标 → 生成子任务列表(planning)
2. 逐个执行子任务(execution)
   ├── 读文件 → 理解现状
   ├── 写代码 → 用 LLM 生成修改
   └── 跑测试 → 验证结果
3. 失败自动重试(最多 3 次,注入反馈)
4. 记录到 SharedMemory + Metrics
5. ContextManager 自动 compaction(&lt;200K 有效窗口)
6. 下次参考历史经验 → 越干越熟

测试策略

148 个测试,覆盖了所有边界情况。

写测试时遵循的几条规则:

  1. Mock 所有的 LLM/API 调用 — 测试不应该联网
  2. 每行代码都有测试覆盖 — 覆盖率 > 95%
  3. 集成测试验证全流程 — coordinator + queue + team + shared_memory
  4. TDD — 先写测试,后写代码

项目统计

指标数值
代码模块16 个 (src/baby_harness/)
Python 行数~1500 行
外部依赖Pydantic 2.x(仅有的运行时依赖)
测试文件15 个
测试用例148 passed, 1 skipped
Lintruff clean
Git commits25+
项目形态Baby Harness(框架) + agent-search-mcp(应用)

下一步

Baby Harness 目前专注于单代理自治开发。后面可以做的:

  1. 多代理并行 — 多个 AgentTeam 同时处理不同子任务
  2. Web Dashboard — 可视化任务状态、指标趋势
  3. 更多 Executor — OpenAI、Claude API 直连
  4. CI/CD 集成 — PR 自动跑 goal.py validation

但这些都是"需要的时候再做"。Baby Harness 的核心理念是 YAGNI(You Ain't Gonna Need It)——现在够用,就不加功能。


后记

Baby Harness 从开始到写完,大约花了 3 个小时。不是因为我厉害,而是因为:

  1. 简单的问题不需要复杂的方案 — 一个自治循环,不需要 Graph、Vector DB、复杂编排
  2. TDD 让设计更清晰 — 测试先写,接口自然就是干净的
  3. 零依赖减少了 90% 的麻烦 — 没有版本冲突、没有 API 变更、没有安装故障

如果你也想做类似的东西,建议从最简版本开始——一个 while 循环 + 一个 LLM 调用。然后慢慢加:重试、记忆、缓存、上下文管理。不要一开始就想做"完美的 AI Agent 框架"。

完成比完美重要。


Baby Harness: github.com/lennney/baby-harness Agent Search MCP: github.com/lennney/agent-search-mcp