从零构建 AI Coding Agent：Hotaru Code

人与人之间最大的差异不是智力，而是他们在社会与环境中积累的上下文。如果一个人完全指导另一个人的上下文（人际关系、项目资料、遇到的各种事情等等），那么这个人大概率也可以替代了另一个人的大部分工作。高中时候读费孝通的《乡土中国》，书中有这样的观点：乡下人的行为模式并不是愚蠢，而是适应了他们的环境；城里人的行为模式也不是聪明，让他们去地里辨认庄稼也答不上来。这句话放到数十年后的 AI 时代来看更有深意。AI Agent 也是一样的道理——它的能力不仅仅是由模型决定的，还是由它的上下文（工具、历史等）塑造的。理解和设计好这个上下文，才是构建优秀 AI Agent 的关键。

本文是我开发 Hotaru Code 过程中的技术总结。Hotaru Code 是一个支持 TUI、WebUI 和 CLI 的 AI 编程助手，具备完整的 Agent Loop、工具调用、权限控制、MCP 协议集成和上下文压缩能力。本文将从架构设计到实现细节，完整剖析一个 AI Agent 的全链路。虽然它可能并不成熟，但我希望通过分享这个过程带来一些启发和参考。

为什么要自己造一个？

作为 Claude Code 的重度用户，我很好奇它是如何做到这么优秀的。我经常想：如果自己从零开始构建一个 AI coding agent，会是什么样子？市面上虽然有一些开源的 agent 框架，但它们要么过于简化，难以扩展，要么抽象过于复杂。我决定只用普通的工具和库，自己实现一个专注于 coding 的 agent，来验证我的设计想法。

Hotaru Code 就是这些想法的落地。Python 3.12+，asyncio 全异步，SQLite WAL 持久化，约 2 万行核心代码。

整体架构：分层与解耦

┌─────────────────────────────────────────────────┐
│              Interface Layer                     │
│   TUI (Textual)  │  WebUI (React)  │  CLI Run   │
├─────────────────────────────────────────────────┤
│              Transport Layer                     │
│   SDKContext → FastAPI Server (SSE + WebSocket)  │
├─────────────────────────────────────────────────┤
│           Session Orchestration                  │
│   SessionPrompt → Processor → TurnRunner         │
│   ToolExecutor → Compaction                      │
├─────────────────────────────────────────────────┤
│              LLM Adapter                         │
│   ProviderTransform → AnthropicSDK / OpenAISDK   │
├─────────────────────────────────────────────────┤
│           Tool & Extension Layer                 │
│   ToolRegistry │ Permission │ MCP │ Skill        │
├─────────────────────────────────────────────────┤
│            Core Infrastructure                   │
│   ConfigManager │ Event Bus │ SQLite Storage     │
└─────────────────────────────────────────────────┘

核心设计原则：上层依赖下层，同层之间通过 Event Bus 通信。Session 层不知道自己跑在 TUI 还是 WebUI 里，Tool 层不关心 LLM 返回的是 Anthropic 还是 OpenAI 格式。这种解耦让每一层都可以独立测试和替换。

三端合一：一个 SessionPrompt 统治一切

三个入口最终汇聚到同一个函数：

TUI:  InputWidget → SessionScreen → SessionPrompt.prompt()
WebUI: POST /v1/sessions/{id}/messages → SessionService → SessionPrompt.prompt()
CLI:  hotaru run "msg" → run_command() → SessionPrompt.prompt()

这是经典的门面模式（Facade）。SessionPrompt 是整个系统的编排中心，负责：

调用 prepare_prompt_context() 解析 provider/model、创建或恢复 session、构建系统提示词
构造 SessionProcessor 并启动 agent loop
循环结束后检查上下文溢出、生成标题、持久化消息

TUI 和 WebUI 的差异仅在传输层：TUI 通过 SDKContext 封装的 HTTP client 与内嵌的 FastAPI server 通信，WebUI 直接走 HTTP/SSE。这意味着 TUI 本质上也是一个”前端”，和 React WebUI 地位平等。

这个设计的好处是显而易见的——任何新增的前端（比如未来的 VS Code 插件）只需要对接 REST API，不需要碰任何业务逻辑。

Agent Loop：LLM 驱动的自主循环

Agent Loop 是整个系统的心脏。它不是简单的”发请求-收回复”，而是一个 LLM 自主决策的循环：

# SessionProcessor.process() 的核心逻辑（简化）
while result.status == "continue" and self.turn < self.max_turns:
    # 1. 准备本轮输入：解析可用工具，构建 StreamInput
    stream_input = self.turnprep.prepare(agent, provider_id, model_id, messages, ruleset)

    # 2. 流式调用 LLM
    async for chunk in self.turnrun.run(stream_input, observer):
        if chunk.type == "text":
            observer.on_text(chunk.text)          # 实时推送文本
        elif chunk.type == "tool_call_end":
            tool_result = await self.tools.execute(chunk.tool_call)  # 执行工具
            messages.append(tool_result)

    # 3. 有工具调用 → 继续循环；没有 → 停止
    if has_tool_calls:
        self.turn += 1
        continue
    result.status = "stop"

这就是 ReAct（Reasoning + Acting）模式的实现。LLM 在每一轮可以选择：

输出文本（思考/回复用户）
调用工具（执行操作）
两者兼有

关键的设计决策：

为什么用流式而不是一次性返回？ 因为工具调用可能需要用户授权。如果等整个响应返回再处理，用户体验会很差——他们看不到 AI 在”思考什么”。流式处理让文本实时显示，工具调用一完成就立即执行。

TurnPreparer 做了什么？ 它不只是简单地把工具列表传给 LLM。它会：

根据当前 agent 的配置过滤可用工具
合并 MCP 工具和内置工具
通过 ToolResolver 生成 OpenAI 格式的 tool definitions
处理 agent 切换（比如从 build agent 切到 plan agent）

LLM 适配层：一套代码接所有 Provider

LLM 世界的现实是：每家 provider 的 API 格式都不一样。Anthropic 用 tool_use/tool_result 块，OpenAI 用 function 调用。Moonshot 把推理过程放在 reasoning_content 字段，Anthropic 用 thinking 块。

ProviderTransform 是解决这个问题的核心：

class ProviderTransform:
    """五步转换管线"""

    # 1. 规范化 tool_call_id（Mistral 要求 9 位字母数字，Claude 不允许特殊字符）
    # 2. 处理 interleaved reasoning（不同 provider 的推理字段名不同）
    # 3. 重映射 provider_options（通过 sdk_key 解析到正确的 SDK）
    # 4. 注入缓存控制（Anthropic: cacheControl, OpenAI: cache_control, Bedrock: cachePoint）
    # 5. 清理空内容（Anthropic 会拒绝空 content 的消息）

转换完成后，根据 api_type 分发到 AnthropicSDK 或 OpenAISDK。两个 SDK wrapper 都输出统一的 StreamChunk：

@dataclass
class StreamChunk:
    type: str       # "text", "tool_call_start", "tool_call_end", "reasoning_delta", ...
    text: str | None
    tool_call: ToolCall | None
    usage: dict[str, int] | None
    stop_reason: str | None

上层代码完全不需要知道底层用的是哪个 provider。这种归一化的代价是 ProviderTransform 本身比较复杂（需要处理各种边界情况），但收益是巨大的——新增一个 provider 只需要在 transform 里加几行映射规则。

SessionRetry 提供指数退避重试，处理 429（rate limit）和 5xx 错误。重试逻辑和业务逻辑完全分离，通过组合而非继承实现。

工具系统：三层叠加的可扩展架构

工具是 Agent 的手和脚。Hotaru 的工具体系分三层：

第一层：内置工具（20+）

文件操作: list, glob, grep, read, edit, write, multiedit, apply_patch
执行环境: bash, task (子 agent 委派)
交互控制: question, todoread, todowrite, plan_enter, plan_exit
网络能力: webfetch, websearch
扩展集成: skill, lsp

每个工具是一个 ToolInfo dataclass，包含四个核心要素：

@dataclass
class ToolInfo:
    id: str                          # 工具标识
    description: str                 # 给 LLM 看的描述
    parameters_type: type[BaseModel] # Pydantic 参数模型（自动校验）
    execute: Callable                # 实际执行函数
    permissions: Callable            # 返回所需权限列表

用 Pydantic 做参数校验是一个关键决策。LLM 生成的 JSON 参数经常有类型错误（比如把数字传成字符串），Pydantic 的 coercion 能自动修正大部分情况，省去了大量手动校验代码。

第二层：MCP 工具

MCP（Model Context Protocol）是 Anthropic 提出的工具扩展协议。Hotaru 实现了完整的 MCP client，支持两种传输：

stdio：启动子进程，通过 stdin/stdout 通信（本地工具）
HTTP/SSE：连接远程 MCP server（支持 OAuth 认证流程）

MCP 工具通过命名空间隔离避免冲突：{client_name}__{tool_name}。比如 filesystem__read_file 和内置的 read 不会混淆。

# MCPManager 的连接策略
async def _connect_remote(self, name, config):
    # 先尝试 streamable HTTP（新协议）
    # 失败则回退到 SSE（旧协议）
    # 遇到 401/403 则触发 OAuth 流程

第三层：自定义工具

用户可以在项目目录的 tool/ 或 tools/ 下放 Python 文件，ToolRegistry 会自动发现并加载。约定优于配置——导出 TOOLS 列表、tool 函数、Tool 类或 register_tools() 函数均可。

权限系统：Human-in-the-Loop 的工程实现

让 AI agent 自动执行代码是危险的。权限系统是安全的最后一道防线。

规则引擎

权限规则由三部分组成：permission（权限类型）、pattern（glob 模式）、action（allow/deny/ask）。

{
  "permission": {
    "bash": "ask",
    "edit": { "src/**": "allow", "*.env": "deny" },
    "read": { "*.env": "ask", "*.env.example": "allow" }
  }
}

匹配逻辑是 last-match-wins——和 CSS 优先级类似，后面的规则覆盖前面的。这比 first-match 更灵活，因为用户可以先设一个宽泛的 deny，再用更具体的 pattern 开白名单。

规则来源按优先级合并：agent 默认规则 → 用户配置规则 → 运行时已批准规则。

异步授权流程

当规则匹配到 ask 时，系统不会阻塞整个进程，而是：

async def ask(self, *, session_id, permission, patterns, ruleset, ...):
    # 1. 创建 asyncio.Future
    future = asyncio.get_event_loop().create_future()
    self._pending[request_id] = future

    # 2. 通过 Bus 发布 PermissionAsked 事件（UI 层订阅并弹出对话框）
    await Bus.publish(PermissionAsked, PermissionAskedProperties(...))

    # 3. await Future（直到用户回复）
    return await future

用户的回复有三种：

once：仅本次放行
always：记住这个规则（作用域可配置：turn/session/project/persisted）
reject：拒绝，并且拒绝当前 session 所有 pending 的权限请求

always 的实现特别巧妙——它不仅 resolve 当前的 Future，还会扫描所有 pending 请求，自动 resolve 匹配同一 pattern 的其他请求。这意味着如果 AI 连续调用了 5 次 bash，用户只需要批准一次。

Doom Loop 检测

LLM 有时会陷入死循环——反复调用同一个工具，传同样的参数。DoomLoopDetector 通过签名匹配检测这种情况：

# 签名 = tool_name + JSON(sorted_input)
signature = f"{tool_name}:{json.dumps(tool_input, sort_keys=True)}"

# 最近 N 次调用中，如果连续 threshold 次签名相同 → 触发权限询问
recent = self.signatures[-self.threshold:]
if len(set(recent)) == 1:
    await self.permission.ask(permission="doom_loop", ...)

默认阈值是 3 次，滑动窗口 50 次。这不是简单地阻止重复调用（有些场景确实需要重复），而是把决定权交给用户。

上下文压缩：让 Agent 拥有”长期记忆”

LLM 的上下文窗口是有限的。一个复杂任务可能涉及几十轮工具调用，每轮的输入输出都会累积 token。不做压缩，很快就会撞到上限。

Hotaru 的压缩策略分两级：

第一级：裁剪（Prune）—— 零成本

class SessionCompaction:
    PRUNE_MINIMUM = 20_000    # 至少裁剪 20k tokens 才值得动手
    PRUNE_PROTECT = 40_000    # 保护最近 40k tokens 不被裁剪
    PRUNE_PROTECTED_TOOLS = {"skill"}  # 某些工具输出永不裁剪

裁剪的目标是旧的工具输出——它们通常是最大的 token 消耗者（一次 grep 可能返回几千行）。裁剪只是把 ToolPart 的内容标记为 compacted，替换为 "[Old tool result content cleared]"。不需要调用 LLM，零额外成本。

第二级：摘要（Summarize）—— 需要 LLM 调用

如果裁剪后仍然溢出，系统会插入一个 CompactionPart 标记，然后调用专门的 compaction agent 生成对话摘要。后续加载历史时，filter_compacted() 会跳过标记之前的所有消息，只保留摘要。

@classmethod
async def is_overflow(cls, *, tokens: TokenUsage, model: ProcessedModelInfo) -> bool:
    # usable = input_limit - reserved_buffer(默认 20k 或 model.limit.output)
    # 当 total_tokens >= usable 时触发
    usable = model.limit.input - reserved
    return tokens.total >= max(usable, 1)

为什么分两级？因为摘要需要额外的 LLM 调用（成本和延迟），而裁剪是免费的。大多数情况下，裁剪就够了——工具输出占了 token 的大头，清理掉旧的输出通常能释放足够空间。只有在对话本身非常长的时候才需要摘要。

Event Bus：解耦的粘合剂

在一个多端架构中，组件之间的通信是最容易变成意大利面条的地方。Hotaru 用 ContextVar 作用域的事件总线解决这个问题：

# 定义事件（类型安全，Pydantic schema）
class PermissionAskedProperties(BaseModel):
    id: str
    session_id: str
    permission: str
    patterns: list[str]

PermissionAsked = BusEvent.define("permission.asked", PermissionAskedProperties)

# 发布
await Bus.publish(PermissionAsked, PermissionAskedProperties(...))

# 订阅
unsubscribe = Bus.subscribe(PermissionAsked, on_permission_asked)

ContextVar 作用域意味着每个请求/session 有自己独立的 Bus 实例，不会串台。这在 WebUI 场景下至关重要——多个用户同时使用时，A 的权限请求不会弹到 B 的界面上。

事件总线承担了几个关键职责：

权限请求/回复的异步通信
消息更新的实时推送（Session → SSE → WebUI）
MCP 工具变更通知
PTY 输出转发

存储层：SQLite WAL 的务实选择

为什么选 SQLite 而不是 PostgreSQL 或 Redis？

零部署成本：用户 pip install 就能用，不需要额外启动数据库服务
WAL 模式：支持并发读写，读操作不阻塞写操作
足够快：对于单用户/少量并发的场景，SQLite 的性能绰绰有余

存储层通过命名空间路由实现表分离：

# key[0] 决定写入哪张表
"sessions"           → sessions 表
"session_index"      → session_index 表
"messages"           → messages 表
"parts"              → parts 表
"permission_approval" → permission_approval 表
其他                  → kv 表（通用键值存储）

update() 方法使用 BEGIN IMMEDIATE 事务，确保读-改-写的原子性。这比乐观锁简单得多，在 SQLite 的单写者模型下也不会有性能问题。

配置系统：五层合并的优先级链

一个好的配置系统需要同时满足”开箱即用”和”深度定制”。Hotaru 的 ConfigManager 从五个来源加载配置，按优先级从低到高深度合并：

1. ~/.config/hotaru/hotaru.json          ← 全局用户配置
2. hotaru.json / hotaru.jsonc            ← 项目根目录（向上查找）
3. .hotaru/hotaru.json                   ← 项目私有配置（向上查找）
4. HOTARU_CONFIG_CONTENT 环境变量          ← JSON 字符串覆盖
5. 托管配置目录                            ← 最高优先级（企业场景）

配置值支持 {env:VAR} 占位符，在加载时解析为环境变量。这让 API key 可以安全地写在配置文件里而不暴露明文：

{ "options": { "apiKey": "{env:OPENAI_API_KEY}" } }

ConfigManager 使用 ContextVar 实现实例隔离——测试中可以注入不同的配置而不影响全局状态。最终合并结果通过 Pydantic 校验，类型错误在启动时就能发现。

子 Agent 委派：Task Tool 的设计

复杂任务往往需要分而治之。TaskTool 允许 LLM 将子任务委派给专门的 agent：

父 Session (build agent)
  → LLM 决定委派
  → TaskTool(subagent_type="explore", prompt="找到所有 API 路由")
  → 创建子 Session（独立 session_id，同一 project）
  → 子 Processor 运行（受限工具集：grep, glob, read, bash）
  → 子 agent 完成 → 文本结果注入父 Session 的 tool_result

子 agent 有自己独立的工具集和权限规则，运行在隔离的 session 中。这意味着 explore agent 不能调用 edit/write（只读），plan agent 不能执行 bash。隔离既是安全措施，也是让 LLM 更专注的手段——工具越少，LLM 越不容易分心。

WebUI：React + SSE 的实时架构

WebUI 的前端是标准的 React SPA，但通信模式值得一提：

REST API：session CRUD、provider 列表、权限回复等操作
SSE（Server-Sent Events）：消息流、工具执行状态的实时推送
WebSocket：终端（xterm.js）的双向通信

为什么用 SSE 而不是全部走 WebSocket？因为 AI 回复本质上是单向流——服务端推送文本块到客户端。SSE 天然适合这个场景，而且比 WebSocket 更简单（自动重连、HTTP 兼容、不需要心跳）。WebSocket 只用在真正需要双向通信的终端场景。

前端的 hooks 设计也值得一提：

useSession()      // session CRUD
useMessages()     // 消息列表 + 增量更新
useEvents()       // SSE 订阅，驱动整个实时更新
usePermissions()  // 权限对话框的状态管理
usePty()          // 终端生命周期

useEvents 是核心——它订阅 SSE 流，根据事件类型分发到其他 hooks。消息更新是增量的（delta），不是每次都拉全量，这对长对话的性能至关重要。

AppContext：分阶段启动与优雅降级

一个有 20+ 子系统的应用，启动顺序很重要。AppContext 实现了分阶段启动：

Phase A: 绑定 ContextVar（Bus, ConfigManager）
Phase B: 加载配置 + 初始化 SQLite（WAL 模式，自动迁移）
Phase C: 注册内置工具（同步，快速）
Phase D: MCP + LSP 并行初始化（带健康追踪）
    ├── MCP 失败 → critical，回滚启动
    └── LSP 失败 → degraded，继续运行（LSP 是增强功能，不是核心）
Phase E: Skill + Agent 并行发现

关键设计：MCP 和 LSP 并行初始化（asyncio.gather），但健康等级不同。MCP 是工具系统的一部分，失败意味着用户配置的工具不可用，必须报错。LSP 只是代码智能提示的增强，降级运行完全可以接受。

这种分级健康追踪避免了”全有或全无”的脆弱性——系统能在部分子系统故障时继续提供核心功能。

开发中的经验与教训

人类是产品经理和架构师，同时也是 AI 的同事

AI 比地球上的任何人都擅长快速实现一个具体的功能，但它不擅长权衡不同功能之间的关系，做出合理的架构设计，如果不加以指导，AI 很可能会写出一个”大杂烩”式的代码，所有功能都混在一起，难以维护和扩展。作为开发者，我们需要时刻提醒自己：我们不是在写一个脚本，而是在构建一个系统。我们需要告诉 AI 如何正确地开发软件（比如 TDD 模式），实现清晰的模块边界划分、接口定义和设计原则，让它知道每个功能应该放在哪里，如何与其他功能交互。

2. 延迟导入解决循环依赖

session/__init__.py 用了一个巧妙的模式——lazy exports：

_EXPORTS = {
    "SessionPrompt": (".prompting", "SessionPrompt"),
    "SessionProcessor": (".processor", "SessionProcessor"),
    ...
}

def __getattr__(name):
    module_name, attr_name = _EXPORTS[name]
    module = import_module(module_name, __name__)
    value = getattr(module, attr_name)
    globals()[name] = value  # 缓存，下次直接返回
    return value

session 包内部有大量交叉引用（processor 引用 compaction，compaction 引用 session，session 引用 processor）。传统的 from .processor import SessionProcessor 会导致循环导入。lazy export 把导入推迟到实际使用时，彻底解决了这个问题。

3. 不要防御性编程掩盖问题

这是我在 CLAUDE.md 里写的第一条规则，也是开发过程中最深刻的教训。早期版本里到处是 try/except Exception: pass，看起来程序”不会崩溃”，实际上是把 bug 藏起来了。

一个真实的例子：MCP 连接偶尔超时，早期的做法是 catch 住然后返回空工具列表。结果用户发现”工具有时候能用有时候不能用”，排查了很久才发现是超时被吞了。正确的做法是让超时错误冒泡到 AppContext 的健康追踪系统，标记为 degraded 并通知用户。

4. ContextVar 是多租户的利器

Bus 和 ConfigManager 都用 ContextVar 实现实例隔离。这在测试中尤其有用——每个测试用例可以注入自己的 Bus 和 Config，完全不需要 mock。在 WebUI 的多用户场景下，每个请求的 ContextVar 自然隔离，不需要额外的租户管理代码。

5. Pydantic 既是校验器也是文档

工具参数用 Pydantic model 定义，一举三得：

LLM 看到的是自动生成的 JSON Schema（tool definition）
运行时自动校验和类型转换
开发者看到的是带类型标注的 Python 类

不需要手写 JSON Schema，不需要手写校验逻辑，不需要维护文档。一个 Pydantic class 搞定一切。

全链路追踪：一条消息的完整旅程

最后，用一条用户消息的完整生命周期串联所有模块，作为全文的总结：

1. 用户在 WebUI 输入 "帮我重构 utils.py"
   → React ChatView → api.sessions.send() → POST /v1/sessions/{id}/messages

2. FastAPI 路由 → SessionService.send_message() → SessionPrompt.prompt()
   → prepare_prompt_context(): 解析 model=claude-sonnet, agent=build
   → Session.get(): 从 SQLite 加载会话历史
   → SystemPrompt.build_full_prompt(): 拼接系统提示词 + agent 指令

3. SessionProcessor.process(user_message)
   → Turn 1: TurnPreparer 解析 22 个可用工具
   → LLM.stream(): ProviderTransform 转换 → AnthropicSDK 发起 HTTP 流
   → LLM 返回: "我先读一下 utils.py" + tool_call(read, {path: "utils.py"})
   → 文本通过 Bus → SSE → WebUI 实时显示
   → ToolExecutor: Permission.evaluate("read", "utils.py") → allow（命中规则）
   → ReadTool.execute() → 返回文件内容
   → 工具结果追加到 messages → 继续循环

4. Turn 2: LLM 分析代码后返回 tool_call(edit, {path: "utils.py", ...})
   → Permission.evaluate("edit", "utils.py") → ask（需要用户确认）
   → Bus.publish(PermissionAsked) → SSE → WebUI 弹出权限对话框
   → 用户点击 "Always Allow" → Permission.reply(always, scope=session)
   → Future resolved → EditTool.execute() → 文件修改完成
   → 结果追加 → 继续循环

5. Turn 3: LLM 返回纯文本 "重构完成，主要改动是..."
   → 无工具调用 → status="stop" → 退出循环

6. SessionPrompt 收尾:
   → SessionCompaction.is_overflow() → false（token 未溢出）
   → SessionSummary.generate() → 异步生成会话标题
   → Session.update_message() → 持久化到 SQLite
   → Bus.publish(MessageUpdated) → SSE → WebUI 更新完成

从用户敲下回车到看到最终回复，经过了 Interface → Transport → Session → LLM → Tool → Permission → Storage → Bus → SSE 九个层次。每一层都有明确的职责边界，每一层都可以独立理解和测试。

写在最后

构建一个 AI coding agent 的过程，本质上是在回答一个问题：如何让 LLM 安全、可控、高效地与真实世界交互？

Hotaru Code 的答案是：用分层架构隔离关注点，用事件总线解耦通信，用权限系统保障安全，用 Provider Transform 屏蔽差异；

最最重要的还是 Context Engineering。Context Engineering 是构建 AI 应用的核心。通过合理运用 Skills、MCP、Subagents，并遵循工具设计和评估的最佳实践，才能构建真正生产可用的 Agent 系统。

代码开源在 GitHub，欢迎 star 和 PR。

技术栈：Python 3.12+ / asyncio / FastAPI / Textual / React / TypeScript / SQLite / MCP Protocol 项目地址：GitHub - hotaru-code

记录从零构建 AI Coding Agent：Hotaru Code