-
Notifications
You must be signed in to change notification settings - Fork 163
11_Tech_Agent_Design
本篇系统性记录 WyckoffAgent CLI Agent 的工程设计。如果你想理解"一个能调用工具、能记住上下文、能后台跑任务的对话 Agent 是怎么搭出来的"，这篇就是完整答案。
用户输入 → 记忆召回 → 系统提示注入 → Provider 流式调用 → chunk 消费
↘ Skill 路由 ↓
┌── text_delta → 渲染到终端
├── tool_calls → 工具执行 → 结果回灌 → 下一轮
├── thinking_delta → 推理过程展示
└── usage → Token 统计
↓
上下文过长 → 压缩 → 记忆落盘
↓
最终回复 → 存入聊天日志
核心运行时：cli/runtime.py → AgentRuntime
用户不想等 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 |
工具失败 | 显示 ✗ 工具名 错误信息 |
model_start |
多轮工具调用间的新一轮模型推理开始 | 重新启动 spinner（"思考中"） |
compaction |
上下文被压缩 | 显示 📦 上下文已压缩 |
retry |
Loop Guard 触发重试 | 显示 ⚠ 警告 |
usage |
Token 用量 | 更新底部统计 |
done |
最终回复 | Markdown 渲染 + 耗时统计 |
- 文字按行缓冲（
_stream_line_buf），遇到换行符才刷新到屏幕 - 旋转动画用 Braille 字符
⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏，80ms 切换一帧 - 如果模型先输出了一段文字然后又发了 tool_calls，已渲染的文字会被擦除（
_pop_lines()）——因为那段文字只是"我打算这样做"的计划，不是最终答案 - 最终回复用 Rich Markdown 渲染，支持表格、代码块、列表等格式
Agent 的工具系统要解决三个问题：
- 告诉模型"你有哪些工具可用"（Schema）
- 模型说"我要调 X 工具"时，实际执行对应的函数（Dispatch）
- 某些危险工具需要用户确认才能执行（Gate）
工具 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）。
每个工具还有运行时行为标记：
@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集合 - 修改后执行 — 可以编辑命令或路径
- 不允许 — 拒绝执行，返回错误给模型
大模型有上下文窗口限制。用户聊久了，消息列表会超出窗口。直接截断会丢失关键信息。
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——防止出现"有工具结果但找不到对应调用"的孤儿消息。
让 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) 在每次用户发消息前执行：
- 从用户消息中提取股票代码（6 位数字正则）
- 提取中文 2-gram 关键词（去停用词，最多 5 个）
-
混合检索：
- FTS5 全文搜索（权重 1.0）
- 股票代码精确匹配（权重 0.85）
- 关键词 LIKE 匹配（权重 0.6）
-
时间衰减：
score × 2^(-天数/30)，半衰期 30 天。preference 类型不衰减 - 取 Top 8 条记忆 + Top 5 条 preference，注入系统提示
# 用户偏好
- 不做创业板，风险承受偏保守
- 持仓不超过5只
# 历史记忆
- [2026-04-20] 分析过600519，认为处于吸筹末期
- [2026-04-18] 用户对尾盘策略表示满意
每种类型有上限（preference 50 条，stock_opinion/decision 30 条，market_view 20 条），超出自动裁剪最旧的。90 天以上的非 preference 记忆自动清理。
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 解决这个问题：
expectation = resolve_turn_expectation(messages) # 推断本轮是否有必须调用的工具
if missing_required_tool(expectation, used_tools):
inject retry message # "不要重复计划，直接调用 XXX 工具"
continue # 强制下一轮最多重试 2 次。仍失败则保留文本 + 显示警告。
如果同一个工具被用相同参数调了 3 次以上，说明模型陷入了死循环。检测方式：
- 精确匹配：
(tool_name, args_hash)在最近 6 次调用中出现 ≥ 3 次 - 模糊匹配：参数的 3-gram Jaccard 相似度 ≥ 0.8
触发后直接返回错误，中断循环。
同一轮模型可能发多个 tool_calls。系统按 concurrency_safe 标记分组：
- 安全的工具并发跑（ThreadPoolExecutor）
- 不安全的串行跑
这大幅加速了"帮我看看 000001、600519、300750"这种批量请求。
用户可能用 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 | 文件 | 覆盖模型 |
|---|---|---|
| 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 |
多模型自动切换 |
很多第三方模型（DeepSeek、Kimi、Qwen）用 OpenAI 兼容接口但不支持所有参数。Provider 渐进式降级：
- 先用完整参数尝试（stream_options、tool_choice、frequency_penalty）
- 如果报错，去掉不支持的参数重试
- 如果模型用 XML 文本输出 tool_calls（如
<tool_call>...</tool_call>），用正则提取并转为标准格式
FallbackProvider 包装多个 Provider，按优先级排序。当活跃 Provider 遇到可重试错误（限流、超时、服务端错误）时，自动切换到下一个。
TUI 会显示黄色 ⚡ 提示告知用户发生了 fallback。
Web 端所有 Provider 统一走 @ai-sdk/openai + 自定义 reasoningFetch，但有两处针对特定模型的兼容适配：
OpenAI 的 structured outputs（strict: true）对 tool schema 有额外约束：
- 不允许
.optional()— 必须用.nullable()声明可为空的字段 - 所有字段必须显式存在于 schema 中，不能缺省
适配方式：tool parameters 中 .optional() → .nullable()，.optional().default(N) 改为直接 .number().describe('通常N')。
位置：web/apps/web/src/lib/chat-agent.ts → buildTools() 中所有 tool 的 parameters。
DeepSeek V4 开启 thinking mode 后，要求每条 assistant 消息都必须携带 reasoning_content 字段回传，否则后续请求报错。
问题：AI SDK 的 @ai-sdk/openai 不解析 DeepSeek 的 delta.reasoning_content，且流式响应无法用 .json() 提取。
适配方式：模块级 reasoningFetch 拦截 SSE 流，通过 TransformStream 透明解析每个 chunk 中的 delta.reasoning_content，累积后存入 reasoningCache。下次请求时自动注入到对应 assistant 消息体中。
请求 → 注入缓存的 reasoning_content 到 body.messages
→ fetch → SSE 流
→ TransformStream（透传 + 解析 reasoning_content）
→ flush() 时推入 reasoningCache
Cache 为模块级变量（跨轮次共享），"新对话" 时调用 resetReasoningCache() 清空。
位置：web/apps/web/src/lib/chat-agent.ts → reasoningFetch + reasoningCache。
全市场漏斗筛选需要 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— 策略回测
让常用的复合操作一键触发，同时支持用户自定义扩展。
| 命令 | 作用 |
|---|---|
/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。
某些复杂任务需要专注的推理上下文，不应该污染主对话。Sub-Agent 是隔离的小型 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，以灰色斜体显示。
每轮对话会写一份 JSONL 追踪文件到 ~/.wyckoff/scratchpad/，记录：
| 事件 | 内容 |
|---|---|
init |
用户输入、系统提示、工具列表 |
thinking |
模型推理过程 |
tool_result |
每次工具调用的入参和结果 |
compaction |
压缩前后的消息列表 |
final |
最终回复文本 |
error |
异常堆栈 |
敏感信息（api_key、token、password 等）自动脱敏。
用途：排查"模型为什么给了奇怪的回复"——看 scratchpad 就能还原完整决策链路。
在长时间对话中，流式连接可能因为网络中断、供应商侧异常等原因进入 TCP 半开状态（half-open）：本地 socket 仍处于 ESTABLISHED 状态，但对端已不再发送数据。此时 for chunk in stream 会永久阻塞，CLI 表现为"卡死"——无 spinner、无超时、无报错。
传统的 socket.settimeout() 对 HTTPS 流式响应无效（SSL 层在内核缓冲后已完成读操作）。ThreadPoolExecutor 方案会在 pool.shutdown(wait=True) 时死锁——因为僵死线程永远不会完成。
cli/runtime.py → _iter_with_timeout(stream, timeout):
┌──────────────┐ ┌──────────────┐
│ daemon thread │ put → │ Queue │ ← get(timeout=60s) ─── 主线程
│ _producer() │ │ │
└──────────────┘ └──────────────┘
│ │
永久阻塞时 超时时
随进程退出 raise TimeoutError
- 生产者线程（daemon=True）：从流中读取 chunk 放入 Queue。流结束放入 SENTINEL，异常放入 (EXCEPTION, exc) 元组。
-
主线程：
q.get(timeout=60)等待 chunk。超过 60 秒无数据 → 抛出TimeoutError→ 上层捕获后重试。 -
关键设计：daemon 线程在半开连接中永久阻塞不会泄漏——进程退出时自动回收，无需
join()。
60 秒是平衡点：
- 模型思考（thinking）时两个 chunk 间隔通常 < 30s
- 太短（5s）会在模型深度推理时误报超时
- 太长（300s）用户等太久才能发现连接已断
超时后 run_stream() 会向 TUI 发送 retry 事件，用户看到"⚠ 模型响应超时"提示。AgentRuntime 会自动重试该轮（不丢失上下文），最多重试 2 次。
在 CLI Agent 中，大部分对话轮次是简单的闲聊或确认（"好的"、"继续"、"什么意思"），不需要调用最贵的模型。如果能自动识别复杂度并路由到合适的模型，可节省 50%+ 的 API 费用。
cli/model_router.py → RoutingProvider:
用户消息 → classify_turn() → heavy / light
↓
RoutingProvider._select(messages)
↓
heavy_provider.chat_stream() 或 light_provider.chat_stream()
RoutingProvider 实现了 LLMProvider 接口，对 AgentRuntime 完全透明——运行时不知道底层在切换模型。
| 判定为 heavy | 条件 |
|---|---|
| 多工具结果待合成 | 最近消息中 tool_result 数量 ≥ 2 |
| 含分析关键词 | 匹配正则：分析/诊断/策略/研报/决策/回测/漏斗/筛选/持仓/复盘/analyze/diagnose/... |
| 长消息 | 用户消息 > 80 字符 |
| 无消息 | 保守默认 |
其余情况判定为 light。默认保守——宁可不降级，也不用小模型处理复杂任务。
wyckoff model light <model_id> # 设置 light 模型
wyckoff model list # 查看模型列表，💡 标记当前 light 模型配置存储在 ~/.wyckoff/wyckoff.json 的 "light" 字段。如果未配置 light 模型，RoutingProvider 不会创建——始终使用主模型。
状态栏通过 RoutingProvider.tier_label 属性显示当前轮次使用的模型和 tier 标记：
- 🧠 heavy 模型名（复杂任务）
- ⚡ light 模型名（简单任务）
当 Agent 通过 exec_command 工具执行 screen（Linux 终端复用器）命令时，screen 会立即 fork 出一个后台守护进程并返回 exit code 0。subprocess.run() 正确捕获到退出码，但有两个隐患：
-
无有用输出：
screen的 stdout/stderr 为空——实际工作在分离的 pty 中进行 - 进程生命周期脱钩：子进程已退出，但 screen session 仍在后台运行，Agent 无法追踪其状态
模型执行 exec_command("screen -dmS my_task ./long_script.sh") → 工具返回 {returncode: 0, stdout: "", stderr: ""} → 模型以为命令已完成 → 实际上脚本仍在后台运行。
对于这类"启动后分离"的命令，通过工具描述引导模型采用正确模式：
-
优先使用后台工具：将长时间运行的任务设计为
background=True的专属工具（如screen_stocks），由系统管理生命周期 -
检查 session 状态：如果必须用 screen，后续用
exec_command("screen -ls")查看 session 是否还在 -
输出重定向：引导模型将输出写入文件（
screen -dmS x bash -c './script.sh > /tmp/out.log 2>&1'），后续读取日志获取结果
exec_command 的 timeout 参数（最大 120s）只能保护同步执行的命令。对于会 daemonize 的命令（screen、nohup、systemd-run 等），timeout 无法覆盖——命令在超时前就已"成功退出"。因此系统设计上应尽量用专属后台工具替代通用命令执行。
| 原则 | 体现 |
|---|---|
| 流式优先 | 第一个字 200ms 内出现，用户不干等 |
| Loop 约束模型 | 模型可以犯错，但状态机会纠偏 |
| 后台不阻塞 | 重计算不卡对话，完成后自动通知 |
| 记忆有衰减 | 30 天半衰期，偏好永不过期 |
| Provider 无关 | 切换模型不需要改任何业务代码 |
| 危险需确认 | 改持仓、执行命令、写文件必须用户点头 |
| 连接不信任 | 流式读取加超时，半开连接不会卡死进程 |
| 成本感知 | 简单对话自动降级到便宜模型，复杂任务走强模型 |
| 可追溯 | Scratchpad 记录完整决策链路 |
- Home
- 01_Product_Overview
- 02_Finance_Wyckoff_Method
- 03_Finance_Quantitative_Metrics
- 04_Finance_Sector_Rotation_Regime
- 05_Finance_Risk_Management
- 06_Backtest_Methodology
- 07_Backtest_Simple_vs_Compound
- 08_Research_Strategy_Decay
- 09_Tech_Architecture
- 10_Tech_LLM_RAG_Integration
- 12_Tech_Actions_Operations
- 13_Tech_Python_Engineering