ai-coding-assistant/docs/plans/2026-05-12-ai-coding-assistant-design.md

AI Coding Assistant 设计文档

日期：2026-05-12 状态：已批准（2026-05-12）

1. 背景与目标

构建一个可扩展的 AI 编程助手，具备以下核心能力：

• 统一多 LLM 提供者 API（OpenAI 兼容 + 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+	asyncio，type 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 抽象

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 server，POST 接收请求，SSE 推送事件，WS 双向通信。

3.4 LLM Provider 抽象

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）

-- 会话表
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 技能系统

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	单元测试 + 集成测试	pytest，stdio 用 subprocess 测试
上下文管理	单元测试	pytest，SQLite 用内存数据库
技能系统	单元测试	pytest，动态加载用 importlib
工具	单元测试	pytest
TypeScript 前端	单元测试	vitest
Web Components	组件测试	vitest + @open-wc/testing
端到端	集成测试	pytest + Playwright（Web）/ subprocess（TUI）

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/