前言

随着 Claude Code、Cursor、Devin 等 AI 编程工具的兴起，“AI Agent + 编程工作台” 成为最热门的 AI 应用方向之一。

这篇文章将完整拆解 Code Agent 项目 —— 从架构设计、核心模块实现到前后端联调，带你看清一个真正可用的 LLM Agent 系统是怎么做出来的。

项目地址： github.com/myh-1302/code_agent

技术栈： Python · Flask · Socket.IO · React 18 · TypeScript ·deepseek api

---

一、为什么要自己造轮子？

市面上已有很多 AI 编程工具，为什么还要自己做？

原因很简单：只有自己造过，才真正懂得其中的难点。

通过构建 Code Agent，我深刻理解了以下问题的工程解法：

• LLM 流式输出如何在前端实时渲染？
• Tool Use（工具调用）的完整生命周期如何管理？
• 多 Agent 协作时，状态同步如何保证一致性？
• 长对话下，上下文 Token 超限怎么处理？
• 如何在 Agent 修改文件时保障安全回滚？

---

二、整体架构设计

┌─────────────────────────────────────────────────┐
│                   Frontend (React 18 + TS)       │
│   文件树  │  对话流 + 工具卡片  │  任务看板/Todo  │
└──────────────────────┬──────────────────────────┘
                       │ WebSocket (Socket.IO)
┌──────────────────────▼──────────────────────────┐
│              api_server.py (Flask + SocketIO)    │
│          RESTful API  +  实时事件推送             │
└──────────────────────┬──────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────┐
│                  Agent Core                      │
│  loop.py (主循环)  │  subagent.py (子 Agent)     │
└──────────────────────┬──────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────┐
│               Components (功能组件)               │
│  memory · safety · todo · task · team · compactor│
└──────────────────────┬──────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────┐
│                  Tools (工具层)                   │
│  10+ 工具：记忆/快照/任务/Todo/团队/Skill...       │
└─────────────────────────────────────────────────┘

四层清晰分离：表现层 → 通信层 → Agent 核心 → 工具层，每层职责单一，易于独立测试和扩展。

---

三、核心模块详解

3.1 Agent 主循环（core/loop.py）

这是整个系统最核心的部分。Agent 循环的本质是一个推理 → 行动 → 观察 → 推理的反复迭代过程：

# 伪代码示意 Agent 主循环
async def agent_loop(messages, tools, callbacks):
    while True:
        # 1. 调用 Claude API 获取推理结果（流式）
        response = await call_claude_api(messages, tools, stream=True)
        
        # 2. 触发事件回调（前端实时更新）
        await callbacks.on_chunk(response.delta)
        
        # 3. 如果是工具调用，执行工具
        if response.stop_reason == "tool_use":
            tool_result = await execute_tool(response.tool_use)
            await callbacks.on_tool_result(tool_result)
            messages.append(tool_result)
            continue  # 带上工具结果继续推理
        
        # 4. 推理完成，退出循环
        if response.stop_reason == "end_turn":
            break

关键设计点：

• 事件回调钩子（callbacks）：每个关键节点（chunk/tool_start/tool_result）都触发回调，解耦 Agent 逻辑与 UI 推送
• 流式中断：通过标志位 + asyncio.CancelledError 优雅处理用户中途终止

3.2 实时通信层（Flask-SocketIO）

前端与后端之间用 Socket.IO 实现双向实时通信，这是整个用户体验的关键。

事件协议设计：

# 服务端推送事件格式
{
    "type": "chunk",           # run_start / chunk / tool_start / tool_result / run_end / error
    "data": {
        "content": "...",      # 文字内容（chunk 时）
        "tool_name": "...",    # 工具名（tool_start 时）
        "input": {...},        # 工具输入
        "output": {...},       # 工具输出
        "elapsed_ms": 1234     # 耗时
    }
}

为什么用 SocketIO 而不是 SSE？

Server-Sent Events（SSE）只支持单向推送，而我们还需要客户端发送中断信号（interrupt），SocketIO 的双向通信天然支持这个场景。

3.3 工具调用体系（tools/）

工具层采用注册模式，每个工具继承 BaseTool 并自动注册：

# tools/base.py
class BaseTool:
    name: str
    description: str
    input_schema: dict
    
    async def execute(self, **kwargs) -> dict:
        raise NotImplementedError

# 工具注册表
TOOL_REGISTRY: dict[str, BaseTool] = {}

def register_tool(tool: BaseTool):
    TOOL_REGISTRY[tool.name] = tool

目前实现的 10+ 工具：

工具	功能
memory	长期记忆读写
safety	文件快照创建/恢复
task_board	任务状态管理
todo	Todo 列表同步
team	子 Agent 协作
background	后台任务执行
compress	上下文压缩
skill	Skill 加载执行
worktree	工作目录管理

3.4 上下文压缩（components/compactor.py）

Claude 的 context window 是有限的，长对话必然面临 Token 超限问题。

解决思路：

# 简化的压缩策略
def compact_messages(messages, token_budget):
    # 1. 统计当前 Token 数
    current_tokens = count_tokens(messages)
    
    if current_tokens <= token_budget:
        return messages
    
    # 2. 保留系统消息 + 最近 N 轮对话
    # 3. 对中间历史调用 Claude 生成摘要
    summary = summarize_history(messages[1:-N])
    
    # 4. 用摘要替换中间历史
    return [system_msg, summary_msg] + messages[-N:]

关键挑战： 压缩时要保留工具调用的配对关系（tool_use 和 tool_result 必须成对出现），否则 API 会报错。

3.5 安全快照系统（components/safety_manager.py）

Agent 会修改文件，万一改错了怎么办？这是实际使用中非常真实的痛点。

解决方案：在 Agent 执行写文件操作前，先创建文件快照。

class SafetyManager:
    def create_checkpoint(self, files: list[str], tag: str):
        """创建文件快照"""
        checkpoint = {
            "id": uuid4(),
            "tag": tag,
            "timestamp": datetime.now(),
            "files": {f: read_file(f) for f in files}
        }
        self.checkpoints.append(checkpoint)
    
    def restore(self, checkpoint_id: str):
        """恢复到指定快照"""
        checkpoint = self.get_checkpoint(checkpoint_id)
        for path, content in checkpoint["files"].items():
            write_file(path, content)

3.6 多 Agent 协作（core/subagent.py + components/team_manager.py）

对于复杂任务，单个 Agent 往往力不从心。Code Agent 支持多 Agent 协作：

• 主 Agent 负责任务拆解与调度
• 子 Agent 各自执行具体子任务（如：分析需求 / 写代码 / 写测试）
• 团队管理器 维护各 Agent 状态，前端实时展示

主 Agent
├── 子 Agent A：需求分析
├── 子 Agent B：代码生成  
└── 子 Agent C：测试验证

---

四、前端架构（React 18 + TypeScript）

前端采用三栏布局：

┌──────────┬────────────────────┬──────────────┐
│  文件树   │    对话 + 工具卡片  │  任务看板/Todo│
│ (左侧栏)  │    (主区域·流式)   │  (右侧栏)    │
└──────────┴────────────────────┴──────────────┘

自定义 Hooks 管理 SocketIO 状态：

// hooks/useSocketIO.ts
function useSocketIO() {
  const [messages, setMessages] = useState<Message[]>([]);
  const [agentState, setAgentState] = useState<AgentState>("idle");
  
  useEffect(() => {
    const socket = io("http://localhost:5000");
    
    socket.on("agent_event", (event: AgentEvent) => {
      switch (event.type) {
        case "chunk":
          // 流式追加文字
          appendChunk(event.data.content);
          break;
        case "tool_start":
          // 新增工具调用卡片
          addToolCard(event.data);
          break;
        case "tool_result":
          // 更新工具卡片结果
          updateToolCard(event.data);
          break;
        case "run_end":
          setAgentState("idle");
          break;
      }
    });
    
    return () => socket.disconnect();
  }, []);
  
  return { messages, agentState, sendMessage, interrupt };
}

工具调用卡片是前端最有特色的设计，每次工具调用都以折叠卡片展示：工具名、输入参数、输出结果、耗时，让用户清晰看到 Agent “在做什么”。

---

五、API 接口设计

提供完整的 RESTful API，方便第三方集成：

GET  /api/status              # Agent 状态
POST /api/chat                # 发送消息
POST /api/chat/interrupt      # 中断执行
GET  /api/history             # 对话历史
GET  /api/tasks               # 任务列表
GET  /api/todos               # Todo 列表
GET  /api/team                # 多 Agent 状态
GET  /api/memory/stats        # 记忆统计
POST /api/memory/search       # 记忆搜索
GET  /api/safety/checkpoints  # 快照列表
POST /api/safety/checkpoint   # 创建快照
POST /api/safety/restore      # 恢复快照
GET  /api/files/tree          # 文件树
GET  /api/files/content       # 文件内容

---

六、一键启动

# 克隆项目
git clone https://github.com/myh-1302/code_agent.git
cd code_agent

# 配置 API Key
echo "ANTHROPIC_AUTH_TOKEN=sk-ant-your-key" > .env
echo "MODEL_ID=claude-sonnet-4-20250514" >> .env

# 安装依赖
pip install -r requirements.txt
cd frontend && npm install && cd ..

# 启动
./start.sh
# 访问 http://localhost:5173

---

七、踩坑与经验总结

坑 1：工具调用配对问题

Claude API 要求 tool_use 和 tool_result 必须成对出现在 messages 中，进行上下文压缩时一定要注意保持配对完整性，否则会得到 400 Bad Request。

坑 2：流式中断的状态恢复

用户点击"中断"后，Agent 循环被打断，此时 messages 列表可能处于不完整状态（有 tool_use 但没有 tool_result）。需要在中断处理中补全消息队列，才能让下一次对话正常进行。

坑 3：SocketIO 事件顺序保证

在高并发推送时，SocketIO 事件可能乱序。解决方案是给每个事件加 sequence_id，前端收到后排序再渲染。

坑 4：React 状态更新与 SocketIO 事件的竞态

React 的 useState 是异步更新的，在 SocketIO 回调中直接读取 state 可能拿到旧值。使用 useRef 存储需要即时读取的状态，useState 只用于触发重渲染。

---

八、后续优化方向

• Agent 循环稳定性 — 长对话下工具调用超时重试、流式中断后的状态一致性恢复
• 上下文压缩策略 — 按 token 预算智能裁剪历史，保留关键决策节点而非简单截断
• 前端渲染性能 — 长消息列表虚拟滚动、工具卡片懒加载、SocketIO 事件节流
• 错误恢复增强 — 分级降级策略（重试→回退→提示用户），减少对话中断
• 文件编辑 Diff 预览 — Agent 修改文件前展示 diff，支持逐块接受/拒绝
• 记忆检索优化 — 向量化存储 + 语义搜索，长生命周期项目中精准召回上下文
• 多工作目录并行 — 侧栏多项目标签页，Agent 跨项目引用代码
• 测试自愈 — Agent 发现测试失败后自动分析根因并提议修复

---

总结

构建 Code Agent 让我真正理解了 LLM Agent 系统的复杂性：它不只是调用API 这么简单，而是需要在实时通信、状态管理、工具编排、安全保障、上下文控制等多个维度同时做好工程设计。

如果你也在学习 AI Agent 开发，强烈建议自己动手做一个完整项目，比看再多教程都有收获。

项目开源地址： github.com/myh-1302/code_agent
欢迎 Star、提 Issue 和 PR！

---