docs: add AI Coding Assistant design and P0 implementation plan

Design covers: agent loop, LLM provider abstraction, transport layer,
context management, skill system, tool registry. P0 plan targets the
minimum viable loop: user input → LLM streaming → tool call → output.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
root 2026-05-12 08:56:41 +08:00
parent 20293ffb8e
commit 98d113883e
3 changed files with 515 additions and 0 deletions

View File

@ -0,0 +1,343 @@
# AI Coding Assistant 设计文档
> 日期2026-05-12
> 状态已批准2026-05-12
## 1. 背景与目标
构建一个可扩展的 AI 编程助手,具备以下核心能力:
- 统一多 LLM 提供者 APIOpenAI 兼容 + Anthropic 原生)
- 工具调用、状态管理、上下文压缩、技能系统
- 多种交互模式TUI、RPC、打印模式、Web UI
- 差分渲染、Markdown、xterm.js 渲染引擎
- Web Components 页面,支持渲染 markdown、html、图片等
### 目标
1. **可扩展**LLM 提供者、工具、技能均支持插件式扩展
2. **流式优先**:从 LLM 调用到终端渲染全链路流式
3. **双语言协作**Python 后端agent 逻辑)+ TypeScript 前端(渲染交互)
4. **多模式交互**:同一 agent 支持不同前端接入TUI / Web / RPC
### 非目标
- 不做模型训练 / 微调
- 不做多用户 / 多租户
- 不做云端部署(初期仅本地运行)
- 不做 VS Code / JetBrains 插件集成(后续扩展)
## 2. 技术栈
| 层 | 技术 | 说明 |
|----|------|------|
| Agent 核心 | Python 3.12+ | asynciotype hints |
| LLM 调用 | httpx (async) | 统一 HTTP 客户端 |
| 状态持久化 | SQLite (aiosqlite) | 消息、会话、技能状态 |
| Transport - stdio | asyncio streams | TUI 直接 spawn |
| Transport - SSE/WS | aiohttp + aiohttp-sse | Web 前端接入 |
| TUI | TypeScript + Ink / Blessed | 通过 stdio JSON-RPC 通信 |
| Web UI | TypeScript + Web Components | 通过 SSE/WS 通信 |
| 渲染 | Markdown-it / xterm.js / 差分渲染 | Web Components 封装 |
## 3. 核心架构
### 3.1 Agent Loop
```
┌─────────────────────────────────────────────────────────────┐
│ agentLoop() / agentLoopContinue() │
│ ├── 创建 EventStream │
│ └── 调用 runLoop() / runAgentLoopContinue() │
└─────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────┐
│ runLoop() — 主循环 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 外层 while(true) — 处理 follow-up 消息 │ │
│ │ ┌───────────────────────────────────────────────┐ │ │
│ │ │ 内层 while(hasMoreToolCalls || pendingMsg) │ │ │
│ │ │ 1. streamAssistantResponse() → LLM 调用 │ │ │
│ │ │ 2. executeToolCalls() → 工具执行 │ │ │
│ │ │ 3. prepareNextTurn() → 可切换模型/思考级别 │ │ │
│ │ │ 4. shouldStopAfterTurn() → 决定是否退出 │ │ │
│ │ └───────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
```
### 3.2 消息流转
```
AgentMessage[] (内部格式)
↓ LLMProvider.convert_messages()
Message[] (LLM API 格式)
↓ LLM 调用
AssistantMessage (含 tool_calls)
↓ executeToolCalls()
ToolResultMessage[]
↓ 回到 runLoop() 继续
```
### 3.3 Transport 抽象
```python
class Transport(ABC):
@abstractmethod
async def start(self, handler: RequestHandler) -> None: ...
@abstractmethod
async def send_event(self, event: AgentEvent) -> None: ...
@abstractmethod
async def stop(self) -> None: ...
# RequestHandler = Callable[AgentRequest, Awaitable[None]]
# AgentEvent = 流式事件token / tool_call / tool_result / done / error
```
**stdio 实现**stdin 读取 JSON-RPC 请求stdout 写入 JSON-RPC 响应/通知。
**SSE/WS 实现**aiohttp 启动 HTTP serverPOST 接收请求SSE 推送事件WS 双向通信。
### 3.4 LLM Provider 抽象
```python
class LLMProvider(ABC):
@abstractmethod
async def stream_chat(
self,
messages: list[Message],
tools: list[ToolDef] | None = None,
**kwargs,
) -> AsyncIterator[StreamChunk]: ...
@abstractmethod
def convert_messages(self, agent_messages: list[AgentMessage]) -> list[Message]: ...
@abstractmethod
def convert_tools(self, tools: list[ToolDef]) -> list[dict]: ...
```
**OpenAI 兼容 adapter**:覆盖 OpenAI / Azure / 通义 / 智谱 / DeepSeek 等。
**Anthropic adapter**:原生 Messages API保留 tool_use / extended_thinking 能力。
### 3.5 上下文管理
```
┌──────────────────────────────────────────────────┐
│ LayeredContext │
│ ┌────────────────────────────────────────────┐ │
│ │ System Layer系统提示词不可压缩 │ │
│ ├────────────────────────────────────────────┤ │
│ │ Task Layer当前任务摘要低压缩优先级 │ │
│ ├────────────────────────────────────────────┤ │
│ │ Conversation Layer对话消息可压缩 │ │
│ │ └── SlidingWindow + Summarizer │ │
│ └────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────┘
```
- **滑动窗口**:保留最近 N 轮完整消息
- **摘要压缩**:超出窗口的旧消息通过 LLM 生成摘要,替换原文
- **分层优先级**System > Task > Conversation压缩只作用于 Conversation 层
### 3.6 状态持久化SQLite
```sql
-- 会话表
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
model TEXT,
metadata JSON
);
-- 消息表
CREATE TABLE messages (
id TEXT PRIMARY KEY,
session_id TEXT REFERENCES sessions(id),
role TEXT NOT NULL, -- system / user / assistant / tool
content TEXT,
tool_calls JSON,
tool_call_id TEXT,
token_count INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 上下文摘要表
CREATE TABLE context_summaries (
id TEXT PRIMARY KEY,
session_id TEXT REFERENCES sessions(id),
layer TEXT NOT NULL, -- system / task / conversation
content TEXT NOT NULL,
message_range_start INTEGER,
message_range_end INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 技能状态表
CREATE TABLE skill_states (
session_id TEXT,
skill_name TEXT,
state JSON,
PRIMARY KEY (session_id, skill_name)
);
```
### 3.7 技能系统
```python
class Skill(ABC):
"""技能基类:提示词模板 + 工具绑定 + 触发规则"""
name: str
description: str
trigger: TriggerRule # 自动 / 关键词 / LLM 选择
system_prompt_fragment: str # 注入到系统提示词
tools: list[ToolDef] # 技能关联的工具
@abstractmethod
async def on_activate(self, ctx: SkillContext) -> None: ...
@abstractmethod
async def on_deactivate(self, ctx: SkillContext) -> None: ...
class SkillRegistry:
"""技能注册中心,支持动态加载"""
def __init__(self, skill_dirs: list[Path]):
self._skills: dict[str, Skill] = {}
self._skill_dirs = skill_dirs
async def load_all(self) -> None:
"""扫描目录,动态导入技能模块"""
async def load_skill(self, name: str) -> None:
"""热加载单个技能"""
def get_active_skills(self, context: AgentContext) -> list[Skill]:
"""根据触发规则返回当前应激活的技能"""
```
**动态加载**:从配置的目录扫描 `.py` 文件,每个文件导出 `Skill` 子类实例,运行时可热加载。
### 3.8 Hooks 配置点
| Hook | 时机 | 用途 |
|------|------|------|
| `get_steering_messages` | 每轮开始前 | 注入上下文消息 |
| `get_follow_up_messages` | Agent 停止前 | 检查后续消息 |
| `prepare_next_turn` | 轮次间 | 切换模型/思考级别 |
| `should_stop_after_turn` | 每轮结束 | 决定是否退出 |
| `before_tool_call` | 工具执行前 | 拦截/修改参数 |
| `after_tool_call` | 工具执行后 | 增强/修改结果 |
## 4. 模块边界与数据流
```
┌─────────┐ JSON-RPC/SSE ┌──────────────┐ StreamChunk ┌───────────┐
│ TUI │ ←──────────────→ │ Transport │ ←─────────────→ │ LLM │
│ (TS) │ │ (stdio/WS) │ │ Provider │
└─────────┘ └──────┬───────┘ └───────────┘
┌─────────┐ SSE/WS ┌──────┴───────┐
│ Web UI │ ←──────────────→│ Agent Loop │
│ (TS) │ │ (Python) │
└─────────┘ └──────┬───────┘
┌──────────────┼──────────────┐
│ │ │
┌─────┴─────┐ ┌────┴─────┐ ┌──────┴─────┐
│ Context │ │ Skills │ │ State │
│ Manager │ │ Registry │ │ (SQLite) │
└───────────┘ └────┬─────┘ └────────────┘
┌─────┴─────┐
│ Tools │
│ Registry │
└───────────┘
```
## 5. 错误处理
| 场景 | 策略 |
|------|------|
| LLM API 调用失败 | 指数退避重试(最多 3 次),返回错误事件给前端 |
| 工具执行超时 | 可配置超时(默认 60s超时后返回 timeout 错误结果 |
| 工具执行异常 | 捕获异常,转为 ToolResultMessage(is_error=True) |
| Transport 断连 | stdio: 进程退出SSE/WS: 自动重连 + 会话恢复 |
| 上下文压缩失败 | 回退到简单截断,保留最近 N 条消息 |
| 技能加载失败 | 跳过该技能,记录警告日志,不影响其他技能 |
## 6. 测试策略
| 层 | 测试类型 | 工具 |
|----|----------|------|
| Python 核心 | 单元测试 | pytest + pytest-asyncio |
| LLM Provider | 集成测试mock HTTP | pytest + respx |
| Transport | 单元测试 + 集成测试 | pyteststdio 用 subprocess 测试 |
| 上下文管理 | 单元测试 | pytestSQLite 用内存数据库 |
| 技能系统 | 单元测试 | pytest动态加载用 importlib |
| 工具 | 单元测试 | pytest |
| TypeScript 前端 | 单元测试 | vitest |
| Web Components | 组件测试 | vitest + @open-wc/testing |
| 端到端 | 集成测试 | pytest + PlaywrightWeb/ subprocessTUI |
## 7. 实现优先级MVP
1. **P0 — 核心 agent loop + OpenAI 兼容 provider + stdio transport**
- 跑通最小闭环:用户输入 → LLM 流式响应 → 工具调用 → 输出
2. **P1 — Anthropic provider + SSE/WS transport + SQLite 状态**
- 多 provider 支持 + Web 接入 + 会话持久化
3. **P2 — 上下文压缩 + 技能系统 + Web Components 基础渲染**
- 长对话支持 + 可扩展能力 + 前端渲染
4. **P3 — TUI + 差分渲染 + xterm.js + 高级渲染**
- 完整交互体验
## 8. 项目结构
```
ai-coding-assistant/
├── pyproject.toml
├── packages/
│ ├── core/ # Python: agent 核心
│ │ ├── agent/
│ │ │ ├── loop.py # agentLoop / runLoop
│ │ │ ├── messages.py # AgentMessage 消息模型
│ │ │ └── hooks.py # Hook 配置点
│ │ ├── llm/
│ │ │ ├── base.py # LLMProvider 抽象接口
│ │ │ ├── openai_compat.py
│ │ │ └── anthropic.py
│ │ ├── transport/
│ │ │ ├── base.py # Transport 抽象接口
│ │ │ ├── stdio.py
│ │ │ └── sse_ws.py
│ │ ├── context/
│ │ │ ├── manager.py # 分层上下文管理
│ │ │ ├── compressor.py # 滑动窗口 + 摘要压缩
│ │ │ └── state.py # SQLite 状态持久化
│ │ ├── skills/
│ │ │ ├── registry.py # SkillRegistry 动态加载
│ │ │ ├── base.py # Skill 基类
│ │ │ └── builtin/ # 内置技能
│ │ └── tools/
│ │ ├── base.py # Tool 基类
│ │ └── builtin/ # 内置工具
│ └── web/ # TypeScript: 前端
│ ├── package.json
│ ├── src/
│ │ ├── components/ # Web Components
│ │ ├── renderers/ # 差分渲染、Markdown、xterm.js
│ │ ├── tui/ # TUI 模式
│ │ └── rpc/ # RPC 客户端
│ └── ...
├── docs/
│ └── plans/
└── memory/
```

View File

@ -0,0 +1,144 @@
# AI Coding Assistant — P0 实现计划
> 日期2026-05-12
> 基于:`docs/plans/2026-05-12-ai-coding-assistant-design.md`(已批准)
> 范围P0 — 核心 agent loop + OpenAI 兼容 provider + stdio transport
## 目标
跑通最小闭环:用户输入 → LLM 流式响应 → 工具调用 → 输出
## 前置条件
- Python 3.12+ 环境
- 依赖httpx, pytest, pytest-asyncio, respx, pydantic
---
## 任务清单
### Batch 1项目脚手架 + 消息模型 + LLM Provider 抽象
#### T01: 初始化项目结构
- **修改目标**:创建 Python 包结构、pyproject.toml、基础配置
- **涉及文件**
- `pyproject.toml` — 项目元数据、依赖、pytest 配置
- `packages/core/__init__.py`
- `packages/core/agent/__init__.py`
- `packages/core/llm/__init__.py`
- `packages/core/transport/__init__.py`
- `packages/core/context/__init__.py`
- `packages/core/skills/__init__.py`
- `packages/core/tools/__init__.py`
- **验证方式**`pip install -e .` 成功,`python -c "import packages.core"` 无报错
- **完成判定**:包可安装、可导入
#### T02: 实现消息模型messages.py
- **修改目标**:定义 Agent 内部消息类型和 LLM API 消息类型
- **涉及文件**
- `packages/core/agent/messages.py` — AgentMessage, UserMessage, AssistantMessage, ToolResultMessage, SystemMessage
- `tests/core/agent/test_messages.py` — 消息模型序列化/反序列化测试
- **验证方式**`pytest tests/core/agent/test_messages.py` 通过
- **完成判定**:所有消息类型可正确创建、序列化为 dict、从 dict 反序列化
#### T03: 实现 LLM Provider 抽象接口llm/base.py
- **修改目标**:定义 LLMProvider ABC、StreamChunk、ToolDef 类型
- **涉及文件**
- `packages/core/llm/base.py` — LLMProvider, StreamChunk, ToolDef, Message 等类型
- `tests/core/llm/test_base.py` — 类型构造测试
- **验证方式**`pytest tests/core/llm/test_base.py` 通过
- **完成判定**:抽象接口可被正确继承,类型可正确构造
---
### Batch 2OpenAI 兼容 Provider + 工具系统基础
#### T04: 实现 OpenAI 兼容 Provider
- **修改目标**:实现 stream_chat、convert_messages、convert_tools
- **涉及文件**
- `packages/core/llm/openai_compat.py` — OpenAICompatProvider
- `tests/core/llm/test_openai_compat.py` — 使用 respx mock HTTP 测试
- **验证方式**`pytest tests/core/llm/test_openai_compat.py` 通过
- **完成判定**
- 流式调用返回正确的 StreamChunk 序列content + tool_call + done
- AgentMessage → OpenAI Message 转换正确
- ToolDef → OpenAI tool 格式转换正确
- 错误重试逻辑(指数退避,最多 3 次)
#### T05: 实现工具系统基础tools/base.py
- **修改目标**:定义 Tool 基类、ToolRegistry、ToolResult
- **涉及文件**
- `packages/core/tools/base.py` — Tool ABC, ToolRegistry, ToolResult
- `packages/core/tools/builtin/echo.py` — 内置 echo 工具(用于测试)
- `tests/core/tools/test_base.py` — 注册、查找、执行测试
- **验证方式**`pytest tests/core/tools/test_base.py` 通过
- **完成判定**:工具可注册、可按名查找、可执行并返回 ToolResult
---
### Batch 3Agent Loop + Transport + 端到端闭环
#### T06: 实现 Agent Loop 核心agent/loop.py
- **修改目标**:实现 runLoop 主循环(流式响应 + 工具调用 + follow-up
- **涉及文件**
- `packages/core/agent/loop.py` — AgentLoop, AgentConfig, runLoop
- `packages/core/agent/hooks.py` — Hook 定义P0 仅定义接口,默认实现)
- `tests/core/agent/test_loop.py` — 使用 mock provider 测试循环逻辑
- **验证方式**`pytest tests/core/agent/test_loop.py` 通过
- **完成判定**
- 单轮对话:用户输入 → LLM 响应 → 结束
- 多轮工具调用LLM 返回 tool_call → 执行工具 → 结果回传 → 继续
- follow-up 消息处理
- 错误处理LLM 调用失败返回错误事件
#### T07: 实现 stdio Transport
- **修改目标**:实现 Transport ABC + stdio JSON-RPC 传输
- **涉及文件**
- `packages/core/transport/base.py` — Transport ABC, AgentEvent, AgentRequest
- `packages/core/transport/stdio.py` — StdioTransport
- `tests/core/transport/test_stdio.py` — subprocess 集成测试
- **验证方式**`pytest tests/core/transport/test_stdio.py` 通过
- **完成判定**
- stdin 接收 JSON-RPC 请求
- stdout 输出 JSON-RPC 响应/通知
- 事件流式输出token / tool_call / tool_result / done / error
#### T08: 实现入口 CLI + 端到端测试
- **修改目标**:创建 CLI 入口,串联所有模块,跑通最小闭环
- **涉及文件**
- `packages/core/__main__.py` — CLI 入口
- `packages/core/cli.py` — 参数解析、配置加载
- `tests/core/test_e2e.py` — 端到端测试mock LLM server
- **验证方式**
- `pytest tests/core/test_e2e.py` 通过
- `python -m packages.core --help` 输出帮助信息
- 手动测试:`echo '{"method":"chat","params":{"message":"hello"}}' | python -m packages.core --provider openai-compat --model gpt-4o` 可获得流式响应
- **完成判定**
- CLI 可启动
- 端到端闭环:输入 → LLM 流式输出 → 工具调用 → 结果输出
- 优雅退出Ctrl+C / EOF
---
## 依赖关系
```
T01 → T02 → T03 → T04 → T06
↘ T05 → T06
T03 → T07
T06 + T07 → T08
```
## 风险与待确认项
1. **OpenAI API Key**:测试需要 mock不需要真实 key手动测试需要用户提供
2. **stdio JSON-RPC 协议细节**:具体消息格式需要在 T07 实现时细化
3. **Python 包命名**`packages.core` 是否合适?考虑改为 `ai_coding_assistant` 或更短的名字

28
memory/2026-05-12.md Normal file
View File

@ -0,0 +1,28 @@
# 2026-05-12
## 决策
- **技术栈**Python 后端 (agent) + TypeScript 前端 (TUI/Web),双语言架构
- 理由Python 生态适合 agent 逻辑和 LLM 交互TypeScript 天然适配 Web Components / xterm.js
- 影响:需要统一 Transport 抽象层桥接两语言
- **Transport 方案**:统一 Transport 接口 + stdio(SSE/WS) 双实现
- 理由TUI 直接 spawn 子进程最简单Web 需要 SSE/WS 流式推送
- 影响agent loop 不感知传输方式,前端可独立选择接入方式
- **LLM Provider**OpenAI 兼容协议 + Anthropic 原生双协议
- 理由OpenAI 兼容覆盖绝大多数提供者Anthropic 原生保留 tool_use/extended_thinking
- 影响:两个 adapterGemini 初期走 OpenAI 兼容层
- **上下文管理**:滑动窗口 + 摘要 + 分层上下文SQLite 持久化
- 理由分层保证关键上下文不被压缩SQLite 支持会话恢复和历史查询
- 影响:需要 aiosqlite 依赖,数据库 schema 设计
- **技能系统**:函数注册 + 提示词模板(方案 C支持动态加载
- 理由:底层工具函数注册 + 上层技能描述编排,灵活可扩展
- 影响:目录扫描 + importlib 动态加载机制
## 纪要
- 完成 AI Coding Assistant 设计文档初稿:`docs/plans/2026-05-12-ai-coding-assistant-design.md`
- 待用户批准设计后进入 Phase 2 (Planning)