Skip to content

11_Tech_Agent_Design

Jump to bottom
youngcan edited this page May 9, 2026 · 48 revisions

Agent 设计专题

本篇系统性记录 WyckoffAgent CLI Agent 的工程设计。如果你想理解"一个能调用工具、能记住上下文、能后台跑任务的对话 Agent 是怎么搭出来的"，这篇就是完整答案。

总览

用户输入 → 记忆召回 → 系统提示注入 → Provider 流式调用 → chunk 消费
         ↘ Skill 路由                                     ↓
                                           ┌── text_delta → 渲染到终端
                                           ├── tool_calls → 工具执行 → 结果回灌 → 下一轮
                                           ├── thinking_delta → 推理过程展示
                                           └── usage → Token 统计
                                                          ↓
                                           上下文过长 → 压缩 → 记忆落盘
                                                          ↓
                                           最终回复 → 存入聊天日志

核心运行时：cli/runtime.py → AgentRuntime

1. 流式输出

为什么要流式

用户不想等 5 秒看到一整坨文字。流式输出让第一个字在 200ms 内出现，体感像"AI 在思考着说话"。

实现机制

Provider 的 chat_stream() 是一个 Python Generator，逐 chunk yield。AgentRuntime.run_stream() 消费 chunk 并翻译为统一的 RuntimeEvent：

事件类型 含义 TUI 渲染方式
text_delta 文字片段 逐行追加到 RichLog
thinking_delta 推理过程片段 静默积累，完成后显示 💭 摘要
tool_calls 模型请求调用工具 清除已流式的文字（因为那只是思考过程）
tool_start 工具开始执行 显示旋转动画
tool_result 工具完成 显示 ✓ 工具名 耗时
tool_error 工具失败 显示 ✗ 工具名 错误信息
compaction 上下文被压缩 显示 📦 上下文已压缩
retry Loop Guard 触发重试 显示 ⚠ 警告
usage Token 用量 更新底部统计
done 最终回复 Markdown 渲染 + 耗时统计

TUI 的流式渲染细节

  • 文字按行缓冲（_stream_line_buf），遇到换行符才刷新到屏幕
  • 旋转动画用 Braille 字符 ⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏，80ms 切换一帧
  • 如果模型先输出了一段文字然后又发了 tool_calls，已渲染的文字会被擦除（_pop_lines()）——因为那段文字只是"我打算这样做"的计划，不是最终答案
  • 最终回复用 Rich Markdown 渲染，支持表格、代码块、列表等格式

2. 工具注册

设计目标

Agent 的工具系统要解决三个问题：

  1. 告诉模型"你有哪些工具可用"（Schema）
  2. 模型说"我要调 X 工具"时，实际执行对应的函数（Dispatch）
  3. 某些危险工具需要用户确认才能执行（Gate）

Schema 定义

工具 Schema 定义在 cli/tools.py 的 TOOL_SCHEMAS 列表，用标准 JSON Schema 格式：

{
    "name": "analyze_stock",
    "description": "单股 Wyckoff 诊断或近期行情查询",
    "parameters": {
        "type": "object",
        "properties": {
            "code": {"type": "string", "description": "股票代码"},
            "mode": {"type": "string", "enum": ["diagnose", "price"]},
        },
        "required": ["code"],
    },
}

每个 Provider 会把这个统一格式翻译成自己的 SDK 格式（Claude 用 input_schema，OpenAI 包一层 {"type": "function"}，Gemini 用 FunctionDeclaration）。

行为元数据（ToolSpec）

每个工具还有运行时行为标记：

@dataclass(frozen=True)
class ToolSpec:
    name: str
    display_name: str         # 中文展示名
    concurrency_safe: bool    # 能否并发执行
    requires_approval: bool   # 是否需要用户确认
    background: bool          # 是否后台执行

执行流程

模型返回 tool_calls
    ↓
ToolRegistry.execute(name, args)
    ↓
需要确认？→ 弹出确认弹窗（允许一次 / 总是允许 / 修改参数 / 拒绝）
    ↓
是后台工具？→ 提交到 BackgroundTaskManager，立即返回 task_id
    ↓
直接调用 → 注入 ToolContext → 执行函数 → 返回结果

并发执行

标记为 concurrency_safe=True 的工具（如 analyze_stock、search_stock_by_name）在同一轮有多个 tool_calls 时，会用 ThreadPoolExecutor(max_workers=5) 并行执行。非安全工具串行执行。

危险工具确认

update_portfolio、exec_command、write_file 标记为 requires_approval=True。

弹窗提供四个选项：

  • 允许一次 — 本次执行，下次还问
  • 本次会话总是允许 — 加入 _always_allowed 集合
  • 修改后执行 — 可以编辑命令或路径
  • 不允许 — 拒绝执行，返回错误给模型

3. 上下文管理与压缩

问题

大模型有上下文窗口限制。用户聊久了，消息列表会超出窗口。直接截断会丢失关键信息。

压缩触发条件

COMPACT_RATIO = 0.25  # 达到上下文窗口的 25% 时触发
TAIL_KEEP = 4         # 保留最近 4 条消息不压缩

不同模型阈值不同：Claude 200K 上下文 → 约 50K token 触发；DeepSeek 64K → 约 16K 触发。

压缩流程

消息列表 = [旧消息...] + [最近4条]
              ↓ head         ↓ tail（保留不动）
              ↓
    ① 记忆落盘：从 head 中提取用户偏好，存入 SQLite
              ↓
    ② 工具结果精简：只保留关键字段（channel、phase、signals、health...）
              ↓
    ③ LLM 摘要：生成 500 字以内的对话摘要
              ↓
    ④ 替换：head → [摘要] + tail

工具结果智能精简

压缩时不是把工具结果全丢掉，而是按工具类型保留关键字段：

  • analyze_stock → 保留 channel、phase、trigger_signals、health
  • portfolio → 保留 diagnostics 或 positions
  • 通用 → 保留 error、message、status、code、name、result

尾部扩展保护

如果 tail 里有引用了某个 tool_call_id 的 tool 消息，但对应的 assistant（发起调用的那条）在 head 里，压缩时会把那条 assistant 拉进 tail——防止出现"有工具结果但找不到对应调用"的孤儿消息。

4. 记忆系统

设计目标

让 Agent 跨会话记住用户的偏好、持仓观点和历史结论，不需要用户每次重新说一遍。

存储

SQLite agent_memory 表 + FTS5 全文索引。

字段 含义
memory_type preference / session / stock_opinion / decision / market_view
content 记忆内容
codes 关联的股票代码（逗号分隔）
created_at 时间戳

记忆写入（两个时机）

会话结束时：对最近 40 条消息调用 LLM 提取结构化记忆。[偏好] 前缀的存为 preference 类型，其余存为 session。

上下文压缩前：在旧消息被压缩丢弃之前，先提取其中的偏好信号存入 SQLite（Memory Flush），防止压缩导致用户偏好丢失。

记忆召回（每轮对话前）

build_memory_context(user_message) 在每次用户发消息前执行：

  1. 从用户消息中提取股票代码（6 位数字正则）
  2. 提取中文 2-gram 关键词（去停用词，最多 5 个）
  3. 混合检索：
    • FTS5 全文搜索（权重 1.0）
    • 股票代码精确匹配（权重 0.85）
    • 关键词 LIKE 匹配（权重 0.6）
  4. 时间衰减：score × 2^(-天数/30)，半衰期 30 天。preference 类型不衰减
  5. 取 Top 8 条记忆 + Top 5 条 preference，注入系统提示

记忆注入格式

# 用户偏好
- 不做创业板，风险承受偏保守
- 持仓不超过5只

# 历史记忆
- [2026-04-20] 分析过600519，认为处于吸筹末期
- [2026-04-18] 用户对尾盘策略表示满意

记忆容量控制

每种类型有上限（preference 50 条，stock_opinion/decision 30 条，market_view 20 条），超出自动裁剪最旧的。90 天以上的非 preference 记忆自动清理。

5. Agent Loop（React 模式）

什么是 React Agent

React（Reasoning + Acting）是当前主流的 Agent 执行模式：模型先推理（Reason），再决定行动（Act），观察结果后继续推理，循环往复直到任务完成。

WyckoffAgent 的 loop 实现在 cli/runtime.py → AgentRuntime.run_stream()。

一轮循环

while True:
    ① 检查是否需要压缩上下文
    ② 调用 provider.chat_stream() 获取模型输出
    ③ 消费所有 chunk
    ④ 如果有 tool_calls:
        - 执行工具（可能并发）
        - 把结果追加到消息列表
        - 继续下一轮（回到①）
    ⑤ 如果没有 tool_calls:
        - 检查 Loop Guard（是否漏调了必需工具）
        - 如果漏了：注入纠偏指令，继续
        - 如果没漏：返回最终文本，结束

Loop Guard（防偷懒机制）

模型有时候会"只说不做"——输出一段计划文字但不调用工具。Loop Guard 解决这个问题：

expectation = resolve_turn_expectation(messages)  # 推断本轮是否有必须调用的工具
if missing_required_tool(expectation, used_tools):
    inject retry message  # "不要重复计划，直接调用 XXX 工具"
    continue              # 强制下一轮

最多重试 2 次。仍失败则保留文本 + 显示警告。

Doom Loop 检测（防死循环）

如果同一个工具被用相同参数调了 3 次以上，说明模型陷入了死循环。检测方式：

  • 精确匹配：(tool_name, args_hash) 在最近 6 次调用中出现 ≥ 3 次
  • 模糊匹配：参数的 3-gram Jaccard 相似度 ≥ 0.8

触发后直接返回错误，中断循环。

并发工具批处理

同一轮模型可能发多个 tool_calls。系统按 concurrency_safe 标记分组：

  • 安全的工具并发跑（ThreadPoolExecutor）
  • 不安全的串行跑

这大幅加速了"帮我看看 000001、600519、300750"这种批量请求。

6. Provider 抽象层

为什么需要抽象

用户可能用 Claude、Gemini、DeepSeek、Qwen 等不同模型。每家的 SDK 接口格式不同，但 Agent Loop 不应该关心底层用的是哪家。

统一接口

class LLMProvider(ABC):
    def chat(self, messages, tools, system_prompt) -> dict: ...
    def chat_stream(self, messages, tools, system_prompt) -> Generator[dict]: ...

消息格式统一

内部用一套规范格式，每个 Provider 负责双向转换：

角色 内部格式
user {"role": "user", "content": "..."}
assistant {"role": "assistant", "content": "...", "tool_calls": [...]}
tool result {"role": "tool", "tool_call_id": "...", "name": "...", "content": "..."}

当前支持的 Provider

Provider 文件 覆盖模型
Claude cli/providers/claude.py Claude 系列
OpenAI cli/providers/openai.py GPT、DeepSeek R1、Qwen、Kimi 等 OpenAI 兼容端点
Gemini cli/providers/gemini.py Gemini 系列
Fallback cli/providers/fallback.py 多模型自动切换

OpenAI Provider 的兼容性处理

很多第三方模型（DeepSeek、Kimi、Qwen）用 OpenAI 兼容接口但不支持所有参数。Provider 渐进式降级：

  1. 先用完整参数尝试（stream_options、tool_choice、frequency_penalty）
  2. 如果报错，去掉不支持的参数重试
  3. 如果模型用 XML 文本输出 tool_calls（如 <tool_call>...</tool_call>），用正则提取并转为标准格式

自动 Fallback

FallbackProvider 包装多个 Provider，按优先级排序。当活跃 Provider 遇到可重试错误（限流、超时、服务端错误）时，自动切换到下一个。

TUI 会显示黄色 ⚡ 提示告知用户发生了 fallback。

7. 后台任务

问题

全市场漏斗筛选需要 2-3 分钟，AI 研报需要 1 分钟。如果同步执行，用户必须干等着，什么都做不了。

解决方案

重计算工具标记为 background=True，提交到后台线程：

模型调用 screen_stocks
    ↓
ToolRegistry 检测到是后台工具
    ↓
提交到 BackgroundTaskManager（daemon thread）
    ↓
立即返回 {"status": "background", "task_id": "..."}
    ↓
模型看到"已提交后台"，继续和用户对话
    ↓
后台线程完成 → 注入一条合成消息通知 Agent → Agent 自动处理结果

后台任务面板

TUI 顶部有一个 BackgroundTaskPanel，每秒轮询一次活跃任务，显示：

⟳ 全市场筛选  L2:六通道甄选  处理中...    [1m23s]

进度上报

后台工具内部可以调用 report_progress(stage, detail, progress) 上报进度。这用 ContextVar 实现线程隔离——只有后台线程的 reporter 会被设置，主线程不受影响。

后台工具列表

  • screen_stocks — 全市场五层漏斗
  • generate_ai_report — AI 三阵营研报
  • generate_strategy_decision — 持仓决策
  • run_backtest — 策略回测

8. Skills 系统

设计目标

让常用的复合操作一键触发，同时支持用户自定义扩展。

内置 Skills

命令 作用
/screen 全市场漏斗筛选
/checkup 持仓健康体检
/report AI 深度研报
/strategy 攻守决策
/backtest 策略回测

实现原理

Skill 本质上就是一个预设的 user message 模板。执行时展开模板，注入到对话中，走完整的 Agent Loop（包括工具调用）。

@dataclass(frozen=True)
class Skill:
    name: str          # "screen"
    description: str   # "/help 里展示的描述"
    prompt: str        # 模板，支持 {user_input} 占位符

用户自定义

在 ~/.wyckoff/skills/ 目录下放 .md 文件即可注册新 Skill：

---
name: dcf
description: DCF 估值分析
---
请对 {user_input} 进行 DCF 估值分析，包括自由现金流预测、折现率假设和目标价估算。

文件名即 Skill 名。用户 Skill 可以覆盖同名内置 Skill。

9. Sub-Agent（子代理）

设计目标

某些复杂任务需要专注的推理上下文，不应该污染主对话。Sub-Agent 是隔离的小型 Agent，有自己的工具子集和系统提示。

三个内置 Sub-Agent

名称 职责 工具子集
Research 信息检索和整理 search_stock_by_name, web_fetch, read_file
Analysis 深度分析 analyze_stock, get_market_overview, query_history
Trading 交易决策 portfolio, update_portfolio, generate_strategy_decision

实现

每个 Sub-Agent 通过 SubAgentToolProxy 限制可用工具，运行自己的 AgentRuntime loop。执行完毕后把结果返回给父 Agent。

进度事件通过 tool_context.on_progress 实时转发到 TUI，以灰色斜体显示。

10. Scratchpad（调试追踪）

每轮对话会写一份 JSONL 追踪文件到 ~/.wyckoff/scratchpad/，记录：

事件 内容
init 用户输入、系统提示、工具列表
thinking 模型推理过程
tool_result 每次工具调用的入参和结果
compaction 压缩前后的消息列表
final 最终回复文本
error 异常堆栈

敏感信息（api_key、token、password 等）自动脱敏。

用途：排查"模型为什么给了奇怪的回复"——看 scratchpad 就能还原完整决策链路。

设计原则总结

原则 体现
流式优先 第一个字 200ms 内出现，用户不干等
Loop 约束模型 模型可以犯错，但状态机会纠偏
后台不阻塞 重计算不卡对话，完成后自动通知
记忆有衰减 30 天半衰期，偏好永不过期
Provider 无关 切换模型不需要改任何业务代码
危险需确认 改持仓、执行命令、写文件必须用户点头
可追溯 Scratchpad 记录完整决策链路

Wyckoff-Analysis Wiki

Agent 工作机制 (9 篇)

Clone this wiki locally