# 从零构建一个类 Claude 的流式 AI Agent：架构设计与实践

## 引言

Claude Code 的发布让很多人看到了 AI Agent 的可能性——它能理解代码、执行命令、读写文件，像一个真正的编程助手。但 Claude 不开源，怎么办？自己造一个。

本文分享我构建一个类 Claude 架构的流式 AI Agent 的完整过程，包括技术选型、架构设计、工具系统、流式输出等核心模块的实现。

## 技术栈

| 模块 | 技术 |
|------|------|
| 后端框架 | FastAPI + WebSocket |
| LLM API | Anthropic SDK（兼容 DeepSeek） |
| 前端框架 | Vue 3 + TypeScript |
| 工具执行 | subprocess + WSL（跨平台） |

## 整体架构

┌─────────────────────────────────────────────────────────┐
│ 前端 (Vue 3) │
│ WebSocket 实时通信 │
└─────────────────────────┬───────────────────────────────┘
│
┌─────────────────────────▼───────────────────────────────┐
│ 后端 (FastAPI) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ Agent │ │ Tools │ │ Event Handler │ │
│ │ Core │ │ Registry │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────────┘ │
└─────────────────────────┬───────────────────────────────┘
│
┌─────────────────────────▼───────────────────────────────┐
│ LLM API │
│ Anthropic / DeepSeek / OpenAI │
└─────────────────────────────────────────────────────────┘

text

## 核心设计

### 1. 模块化分层

将 Agent 拆分为独立的模块，各司其职：

| 模块 | 职责 |
|------|------|
| `StreamingAgent` | 主控逻辑，ReAct 循环调度 |
| `EventHandler` | 解析 LLM 流式事件 |
| `MessageBuilder` | 构建 API 消息格式 |
| `Context` | 管理对话历史 |
| `ToolRegistry` | 工具注册与执行 |

**好处**：每个模块只做一件事，易于测试和扩展。

### 2. 工具系统设计

工具系统是 Agent 的核心能力。我设计了一个可插拔的工具架构：

```python
class BaseTool(ABC):
    name: str
    description: str
    input_schema: dict
    
    @abstractmethod
    async def execute(self, args: dict) -> str:
        pass

内置工具示例：

工具	功能	跨平台实现
bash	执行 Shell 命令	Windows 用 WSL，Linux/Mac 用 bash
read_file	读取文件	路径沙箱 + 异步读取
write_file	写入文件	自动创建父目录
load_skill	加载技能知识	动态加载 SKILL.md

3. 流式事件处理

LLM 的流式响应需要解析多种事件类型：

python

for event in stream:
    if event.type == "content_block_delta":
        # 文本增量 → 逐字推送前端
        await send_callback({"delta": event.delta.text})
    elif event.type == "content_block_start":
        # 工具调用开始
    elif event.type == "tool_use":
        # 完整工具调用

关键点：每条消息后加 await asyncio.sleep(0) 主动让出 CPU，让 WebSocket 真正发送数据。

4. 跨平台命令执行

LLM 默认生成 Linux 命令，如何在 Windows 上运行？

python

if platform.system() == "Windows":
    if shutil.which("wsl"):
        # 使用 WSL 执行
        await asyncio.create_subprocess_exec("wsl", "bash", "-c", command)
    else:
        return "Error: Please install WSL"

5. ReAct 循环

Agent 的核心是 ReAct（Reasoning + Acting）循环：

python

async def process_message(self, user_message):
    self.context.add_user_message(user_message)
    
    for _ in range(max_iterations):
        response = await self._call_llm()
        
        if not response.has_tool_calls:
            # 无工具调用，直接结束
            self.context.add_assistant_message(response.content)
            return
        
        # 有工具调用：执行工具，继续循环
        self.context.add_assistant_message(response.content, response.tool_calls)
        for tool in response.tool_calls:
            result = await self._execute_tool(tool.name, tool.input)
            self.context.add_tool_result(tool.id, result)
        # 继续下一轮

踩坑记录

坑 1：前端不流式，等全部完成才显示

原因：send_callback 只是把数据放入缓冲区，没有真正发送。

解决：每次 await send_callback() 后加 await asyncio.sleep(0)，主动让出 CPU。

坑 2：Windows 上执行 Linux 命令失败

原因：subprocess.run 默认用 cmd.exe，不认识 ls、cat。

解决：检测 WSL，用 ["wsl", "bash", "-c", command] 执行。

坑 3：虚拟环境被上传到 GitHub

原因：.gitignore 没有配置虚拟环境目录。

解决：添加 deepL_agent/ 到 .gitignore。

项目结构

text

DeepL-Agent/
├── backend/
│   ├── app/
│   │   ├── agent/           # Agent 核心
│   │   ├── tools/           # 工具系统
│   │   ├── main.py          # WebSocket 服务
│   │   └── skill_loader.py  # 技能加载
│   ├── skills/              # 技能文件
│   └── requirements.txt
├── frontend/
│   └── src/
│       └── ChatWindow.vue
└── README.md

运行方式

bash

# 后端
cd backend
pip install -r requirements.txt
python app/main.py

# 前端
cd frontend
npm install
npm run dev

总结

构建一个 AI Agent 的核心挑战：

1. 流式体验：让用户感知到 Agent 在"思考"

2. 工具扩展：如何设计可插拔的工具系统

3. 跨平台：LLM 生成的命令如何在不同系统上执行

4. 状态管理：多轮对话的上下文维护

这个项目还在持续迭代中，后续计划支持 MCP 协议、长期记忆等功能。

相关链接

• GitHub：https://github.com/aminga123321/DeepL-Agent-MiniClaude-

• 技术栈：FastAPI + Vue 3 + Anthropic SDK