- 其实这个项目几乎是我和我同学Vibe Coding出来的，但是完全依靠Vibe Coding也不行啊，还是要自己学点技术
- 前端时间太忙了，杂七杂八的事一大堆，现在终于处理的差不多了，是时候要学学技术了~~
Vibe Coding是真会把人惯废大概讲一下我做这个项目的心路历程：
- 首先，我是真的心累，各种意义上的心累，我的理想很美好，想做一个Live2D的二次元桌宠，然后让ta有自己的思维，可以帮我干活，可是途中遇到了各种各样的问题，真的是，遇到了一万个问题，真是燃尽我了，做不动了真的，所以后面的想法等放假再说吧...
但其实整体上也算完成了当初想法的90%了吧，但是各种功能的效果真的都很一般...
- 放几张效果图：
- 我打算先从后端结构讲起，首先我不是什么技术大佬，写这个博客也只是为了促进自己的学习罢了，但其实也没人会看我的博客罢了，在这里，我向线下认识我，并且把我的博客盒了出来的朋友们问个好：你们好喵！
所以全网同名还是太容易盒了 参考文献以及项目（感谢各位大佬们）：
(还参考了很多很多项目，后续会慢慢完善的，感谢各位大佬orz)- 1.project-airi Neko大佬和洛灵大佬太强了
- 2.EchoBot 感谢orz
backend
Agent structure （simply）
- 总得来说，我们的架构采用的是如下三层：
请求进来 → L1(判断意图) → L2(角色包装) → L3(执行工作) → 结果出去
↑ │
│ 过程中 │
└──── 消息管道 ───────────────┘
持续推送 quip/表情/状态- 第一层是
router路由，第二层是roleplay，第三层就是类ReAct架构（但是很这个玩意很sb，因为把意图检查放在Layer1，就会很死板，之后可能还是要继续优化,如果把第三层做成完整ReAct架构应该会更好，但是那样有点难调试...）。 其实这样做的目的也很简单：
- 1.减轻上下文腐化现象，给
roleplay和workflow层都留好干净的上下文 - 2.更好的模拟出一个二次元形象
- 3.大幅增强可拓展性
- 4.当然含有很多很多好处...
- 1.减轻上下文腐化现象，给
Layer 0 预处理
整体架构:
POST /chat {"prompt": "帮我写个 calculator.py"}
│
▼
┌───────────────────────────────────────────────────────────────────────┐
│ routing_guard.route(prompt, context) │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ ① detect_intent(prompt) │ │
│ │ │ │
│ │ ┌──────────────────────┐ │ │
│ │ │ _is_empty_or_noise? │──YES──▶ return "unknown" │ │
│ │ └────────┬─────────────┘ │ │
│ │ │NO │ │
│ │ ▼ │ │
│ │ ┌──────────────────────────┐ │ │
│ │ │ detect_run_action() │ │ │
│ │ │ 有UUID + retry/rerun/ │──YES──▶ return "coding" │ │
│ │ │ cancel/inspect 关键词? │ → run.retry/rerun/ │ │
│ │ └────────┬─────────────────┘ cancel/inspect │ │
│ │ │NO │ │
│ │ ▼ │ │
│ │ ┌──────────────────────────────────────────────────────┐ │ │
│ │ │ looks_like_coding_prompt(prompt) │ │ │
│ │ │ │ │ │
│ │ │ ╔══════════════════════════════════════════════════╗ │ │ │
│ │ │ ║ 第1层: 强信号 一票通过 ║ │ │ │
│ │ │ ║ has_file_reference() → .py .js .vue .json... ║ │ │ │
│ │ │ ║ _has_code_structure() → def class import ... ║ │ │ │
│ │ │ ║ 命中任一 → "coding" ║ │ │ │
│ │ │ ╚══════════════════════════════════════════════════╝ │ │ │
│ │ │ │未命中 │ │ │
│ │ │ ▼ │ │ │
│ │ │ ╔══════════════════════════════════════════════════╗ │ │ │
│ │ │ ║ 第2层: 组合判断 (8个信号 → 3条规则) ║ │ │ │
│ │ │ ║ ║ │ │ │
│ │ │ ║ 收集信号: ║ │ │ │
│ │ │ ║ ├─ has_action "写/改/删/运行..." ║ │ │ │
│ │ │ ║ ├─ has_strong_action "写/删/重构..." ║ │ │ │
│ │ │ ║ ├─ has_object "代码/文件/函数..." ║ │ │ │
│ │ │ ║ ├─ has_deliverable "demo/calculator..." ║ │ │ │
│ │ │ ║ ├─ has_path "src/components/..." ║ │ │ │
│ │ │ ║ ├─ has_command "pytest/npm/git..." ║ │ │ │
│ │ │ ║ ├─ has_issue "bug/报错/异常..." ║ │ │ │
│ │ │ ║ └─ has_tech "Python/Vue/React..." ║ │ │ │
│ │ │ ║ ║ │ │ │
│ │ │ ║ 规则A: 报错 + (对象|命令|技术) → coding ║ │ │ │
│ │ │ ║ 规则B: 动作 + (对象|交付物|路径|命令) → coding ║ │ │ │
│ │ │ ║ 规则C: 强动作 + (技术|交付物) → coding ║ │ │ │
│ │ │ ╚══════════════════════════════════════════════════╝ │ │ │
│ │ │ │均未命中 │ │ │
│ │ │ ▼ │ │ │
│ │ │ return "chat" │ │ │
│ │ └──────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
│ ▼ │
│ ┌────────────────┼────────────────┐ │
│ │ │ │ │
│ intent=chat intent=coding intent=unknown │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ for_chat() │ │_route_coding │ │ for_unknown()│ │
│ │ │ │ (prompt, │ │ │ │
│ │ action= │ │ context) │ │ action= │ │
│ │ chat.reply │ └──────┬───────┘ │ final.answer │ │
│ └────────────┘ │ └──────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ ② detect_run_action(prompt) │ │
│ │ 有UUID+控制词? │ │
│ │ YES → run.retry/rerun/cancel/inspect│ │
│ │ NO → _classify_coding_action() │ │
│ └──────────────┬──────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ ③ _classify_coding_action() │ │
│ │ │ │
│ │ Tier 0: 写/删/移/复制 │ │
│ │ → LLM 提取参数 + 生成文件内容 │ │
│ │ │ │
│ │ Tier 1: 测试/搜索 │ │
│ │ → LLM 提取参数 │ │
│ │ │ │
│ │ Tier 2: 读文件 │ │
│ │ → 正则提取路径 → 失败 → LLM 兜底 │ │
│ │ │ │
│ │ Tier 3: 列目录 │ │
│ │ → 直接返回 │ │
│ │ │ │
│ │ 默认: run.create (代码生成) │ │
│ └──────────────┬──────────────────────┘ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ ④ _extract_workspace_params_via_llm()│ │
│ │ │ │
│ │ system_prompt = EXTRACTION_PROMPT │ │
│ │ temperature = 0.1 │ │
│ │ max_tokens = 4000 │ │
│ │ │ │
│ │ LLM 返回 JSON: │ │
│ │ {"rel_path":"calc.py", │ │
│ │ "content":"def add(a,b):...", │ │
│ │ "overwrite":false} │ │
│ └──────────────┬──────────────────────┘ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ ⑤ RoutingDecision(frozen=True) │ │
│ │ │ │
│ │ intent = "coding" │ │
│ │ action_name = "workspace.write" │ │
│ │ action_input = {"rel_path": "calculator.py", │ │
│ │ "content": "def add(a,b):\n return a+b", │ │
│ │ "overwrite": false} │ │
│ │ reason = "Coding intent detected." │ │
│ │ │ │
│ │ ⚠️ frozen: 创建后不可修改，L2只能读不能改 │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │ │
└─────────────────────────┼────────────────────────────────────────────┘
│
▼
┌───────────────────────┐
│ 传给 L2 RoleplayAgent │
│ roleplay_agent │
│ .process(decision) │
└───────────────────────┘- 前端
POST请求先送到api/chat_routes.py,由api/chat_routes.py将HTTP请求转换为函数参数，并且在最后后端流程跑完后，再由api/chat_routes.py将函数返回值转换成HTTP相应JSON。 - 大致原理：
HTTP 请求 HTTP 响应
{"prompt":"...","context":null,...} {"ok":true,"intent":"coding",...}
│ ▲
▼ │
┌──────────────────┐ ┌──────────────────┐
│ req: ChatRequest │ │ ChatResponse │
│ (FastAPI 自动解析)│ │ (FastAPI 自动序列化)│
└────────┬─────────┘ └────────▲─────────┘
│ │
▼ │
req.prompt, req.context, req.session_id │
│ │
▼ │
normalize_prompt_and_context() │
│ │
▼ │
generate_chat_response() ──────────→ build_chat_response()- 我（codex）定义的解析规则为：
class ChatRequest(BaseModel):
prompt: str = Field(..., min_length=1, description="用户输入")
context: str | None = Field(default=None, description="对话上下文")
session_id: str | None = Field(default=None, description="会话 ID；不传则由后端创建")- 随后由
api/chat_routes.py调用services/chat.py,services/chat.py的功能则是用于按顺序调用其他模块，然后把结果拼成最终回复。 services/chat.py文件结构：
┌─────────────────────────────────────────────┐
│ 上半部分：数据定义 │
│ │
│ class ChatServiceResult ← 最终结果的容器 │
│ │
│ helper 函数: │
│ _apply_scheduled_output │
│ _schedule_coding_run_if_needed │
│ _persist_result_to_conversation │
└─────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ 下半部分：主函数 │
│ │
│ generate_chat_response() │
│ 9 个步骤，串联所有模块 │
└─────────────────────────────────────────────┘- 所以其实后端整体的工作流控制是由
services/chat.py中的generate_chat_response()进行的，class ChatServiceResult则就是我前面说的，将函数返回值转换成响应JSON的结果容器。 generate_chat_response()流程图
generate_chat_response(prompt, context, session_id, schedule_run)
│
├─ 步骤1: 会话管理
├─ 步骤2: 上下文拼接
├─ 步骤3: L1预判 (拿 intent_hint)
├─ 步骤4: 发聊天开始反馈
├─ 步骤5: 文件上下文
├─ 步骤6: L1正式路由
├─ 步骤7: L2+L3执行 (线程池)
├─ 步骤8: Hermes记忆
└─ 步骤9: 构建结果 + 调度 + 持久化步骤1: 会话管理
主要目的是确定会话ID
active_session_id = conversation_store.get_or_create_session_id(session_id)- 前端传了
session_id──→ 复用已有会话，加载历史消息 - 前端没传
session_id──→ 创建新会话，返回新UUID conversation_store操作backend/workspace/conversations/下的 JSON 文件。每个会话一个文件。
步骤2: 上下文拼接
- 用的笨办法，直接拼接字符串
effective_context = conversation_store.build_context(active_session_id, context)
memory_context = hermes_memory.build_context(active_session_id)
if memory_context:
if effective_context:
effective_context = f"{memory_context}\n\n---\n\n{effective_context}"
else:
effective_context = memory_context- 对于处理对话历史
memory_context: build_context(session_id, external_context)去提取记忆数据，随后调用排版引擎函数_build_context_text(messages, metadata, external_context),并按固定顺序拼接三个段落：- 对于处理当前对话记忆
memory_context: - 使用结构化记忆，随后再格式化输出
@dataclass
class MemoryEvent:
turn_index: int # 第几轮
timestamp: str # 时间戳
user_input: str # 用户说了什么（截断200字符）
intent: str # L1 路由判断的意图（chat / coding / file_ops / ...）
action_name: str # 执行了什么动作（run_code / edit_file / ...）
result_summary: str # 执行结果摘要（截断200字符）
ok: bool # 成功还是失败
error: str | None # 错误信息
@dataclass
class MemoryEvent:
turn_index: int # 第几轮
timestamp: str # 时间戳
user_input: str # 用户说了什么（截断200字符）
intent: str # L1 路由判断的意图（chat / coding / file_ops / ...）
action_name: str # 执行了什么动作（run_code / edit_file / ...）
result_summary: str # 执行结果摘要（截断200字符）
ok: bool # 成功还是失败
error: str | None # 错误信息┌──────────────────────────────────────┐
│ ① Client provided context │ ← 前端传的额外上下文（可选）
│ （截断到 2000 字符） │
├──────────────────────────────────────┤
│ ② Compressed earlier summary │ ← 旧消息的压缩摘要
│ covering N older messages │
├──────────────────────────────────────┤
│ ③ Recent stored conversation history │ ← 最近8条消息原文
│ User: xxx │
│ Assistant: xxx │
└──────────────────────────────────────┘最后将两种记忆选择性合并
┌─────────────────┐ ┌─────────────────┐ │ Hermes 长期记忆 │ │ 会话历史消息 │ │ (跨会话,结构化) │ │ (当前会话,N条) │ │ │ │ │ │ "用户讨厌Java │ │ user: 帮我看文件 │ │ 偏好Python │ │ assistant: 好的 │ │ 上次写了fib.py" │ │ user: 改一下 │ └────────┬────────┘ └────────┬────────┘ │ │ └───────────┬───────────┘ ▼ "用户讨厌Java,偏好Python... --- user: 帮我看文件 assistant: 好的 user: 改一下"- 结构简单点，但是感觉很容易造成上下文腐化，并且如果记忆太多，会加大token的消耗，总的来说还是太史了。
- 如果换成现在常用的结构化消息流应该会好一些。
步骤3: L1预判 (拿 intent_hint)
- 这一块貌似有点没必要？
步骤3 + 步骤6：L1 被调用了两次,对比一下：
- 步骤3: 第1次调用 —— 只为了快速知道是chat还是coding
intent_hint = routing_guard.route(prompt, context).intent- 步骤6: 第2次调用 —— 正式路由,带完整上下文
decision = routing_guard.route(prompt, effective_context, file_context)
- 这里大致讲一下三个
context的区别（虽然都是AI给的罢了）：
- 第一次调用L1是为了进行快速的嗅探， 仅靠
prompt和前端的现场提示判断意图，随后注入到上下文中。 - 主要目的就是为了在拿到完整的上下文之前发出一个提示性
quip，输出一个意图提示的quip。但这个我感觉完全没有必要...
步骤4: 发聊天开始反馈
- 写死的一个状态显示，用于开启绿灯闪烁和提示前端气泡，进行一个运行状态的证明
def send_chat_started():
return dispatch_character_event(CHAT_STARTED_EVENT)
CHAT_STARTED_EVENT = CharacterEvent(
quip="我想一下...", # → MessageQueue → 前端气泡
expression="thinking", # → MessageQueue → Live2D 表情
status="running", # → MessageQueue → 绿点闪烁
progress=5,
)步骤5: 文件上下文
- 用于获取同一会话中之前操作过的文件记录，形成文件上下文，为第二次
L1调用做好准备。
file_context = None
from ..storage.file_context_store import file_context_store
stored_fc = file_context_store.get_context(active_session_id)
if stored_fc:
file_context = stored_fc步骤6: L1正式路由
- 这个放在后面讲
步骤7: L2+L3执行 (线程池)
- 一样后面一起讲
步骤8: 记录到Hermes记忆
hermes_memory.record_turn(
session_id=active_session_id,
user_input=prompt,
intent=decision.intent,
action_name=decision.action_name,
result_summary=process_result.response.chat_line[:200], # 截前200字
ok=True,
)大语言模型本身是无状态的，所以要保存记忆，这段代码记录了一次完整的交互轨迹。
- user_input：用户的原始触发条件。
- intent & action_name：Agent 大脑解析出的意图和调用的工具/行为。
- ok=True：该动作执行的最终状态。
- 例如：当 Agent 下一次需要回忆时，这段结构化数据可以让系统知道：“用户上次让我查天气（intent），我调用了天气 API（action_name），并且成功了（ok=True）。”
步骤9: 构建结果 + 调度 + 持久化
L2+L3 线程池返回 process_result
│
│ process_result = ProcessResult(
│ response = RoleplayResponse(chat_line="搞定啦！已创建文件 ⭐", expression="proud", quip="机魂大悦！", ...)
│ work_metadata = {"run_id":"abc-123", "run_action":"create", "stop_reason":"done", ...}
│ )
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ services/chat.py 收尾四步 │
│ │
│ ① Hermes 记忆记录 │
│ hermes_memory.record_turn(session_id, prompt, intent, action, result...) │
│ → 写入结构化日志: [Turn 3] OK | Intent: coding | Action: workspace.write │
│ │
│ ② 构建 ChatServiceResult │
│ 从 ProcessResult 中拆解出: run_id, run_action, chat_line, ok, error │
│ 组装成 ChatServiceResult 对象 │
│ │
│ ③ 调度后台执行 (_schedule_coding_run_if_needed) │
│ 如果 coding + 有 run_id + 有调度器 → 放入后台队列执行 │
│ │
│ ④ 持久化到对话存储 (_persist_result_to_conversation) │
│ conversation_store.append_message(session, "user", prompt) │
│ conversation_store.append_message(session, "assistant", output) │
│ 发 "聊天结束" 信号 (send_chat_done / send_chat_failed) │
│ │
│ return result.attach_session(active_session_id) │
└─────────────────────────────────────────────────────────────────────────────┘- 然后留着后台继续跑调度器，待调度器结束后，通过
WebSocket把结果推送至前端
schedule_run("abc-123")
│
│ FastAPI background_tasks.add_task(execute_run, "abc-123")
│
▼
┌────────────────────────────────────────────────────────────┐
│ execute_run("abc-123") 后台线程 │
│ │
│ 循环: │
│ ├─ 生成代码 → 写入文件 → 执行脚本 │
│ ├─ 成功? → finalize_success │
│ ├─ 失败? → LLM自动修复 → 重试 (最多N次) │
│ └─ 用户取消了? → finalize_cancelled │
│ │
│ 无论如何结束: │
│ ├─ update_run_record(run_id, status="done"/"failed") │
│ │ → 状态写入磁盘 │
│ │ │
│ ├─ send_task_done() / send_task_failed() │
│ │ → 通过 WebSocket 推给前端: "任务完成/失败" │
│ │ │
│ └─ emit_final_run_chat_message() │
│ → 用 Shion 的口吻把执行结果推送到聊天窗口 │
│ → "代码跑完啦！生成了 3 个文件 ⭐" │
│ │
└────────────────────────────────────────────────────────────┘Layer 1: Router 路由门卫
- 不执行任何工作，主要用于意图判断。
在三层架构中的位置
POST /chat { prompt: "帮我写个排序", context: "当前在 app.ts" }
│
▼
┌─────────────────────────────────────────────────┐
│ services/chat.py │
│ │
│ ① 拼接上下文 │
│ ② ★ L1: decision = routing_guard.route(...) │
│ ③ L2: roleplay_agent.process(decision) │
│ ④ L3: work_agent.execute(decision) — L2内部调用 │
│ ⑤ 构建结果 + 持久化 + 返回 │
└─────────────────────────────────────────────────┘L1 被 L2 和 L3 依赖，但它不依赖任何一层。它在调用链的最前面，是一次纯判断。
核心架构
routing_guard.route(prompt, effective_context, file_context)
│
▼
┌───────────────────────────────────────────────────────┐
│ RoutingGuard │
│ │
│ def route(self, prompt, context, file_context): │
│ intent = detect_intent(prompt) ← 引擎① │
│ │
│ if intent == "chat": │
│ → for_chat(prompt, context) ← 工厂② │
│ │
│ if intent == "coding": │
│ → _route_coding(prompt, context) ← 分流③ │
│ │ │
│ ├─ 运行控制? → run.retry/rerun/cancel │
│ │ │
│ └─ 新请求 → _classify_coding_action() │
│ │ ← 分类器④ │
│ ├─ Tier1: write/delete/move/copy │
│ │ → LLM提取参数 │
│ ├─ Tier2: test/search │
│ │ → LLM提取参数 │
│ ├─ Tier3: read │
│ │ → 正则优先，LLM兜底 │
│ ├─ Tier4: list │
│ │ → 直接执行 │
│ └─ Default: run.create │
│ → 兜底 │
│ │
│ if intent == "unknown": │
│ → for_unknown(prompt) ← 兜底⑤ │
│ │
│ 输出: RoutingDecision (frozen, 不可变) │
└───────────────────────────────────────────────────────┘- 整个
L1由五个组件构成：一个意图引擎、一个chat工厂、一个coding分流器、一个四级动作分类器、一个unknown兜底。
引擎①：detect_intent() —— 意图判定
- 这是 L1 的第一次意图匹配。纯关键词 + 正则匹配，不调 LLM，零延迟。
detect_intent(prompt)
│
├─ ① 空消息 / 纯符号？
│ └→ "unknown"
│
├─ ② 包含运行控制动作？(retry / rerun / cancel / inspect)
│ └→ "coding" ← 只要在操作 run，一定是编程场景
│
├─ ③ looks_like_coding_prompt()？
│ └→ "coding"
│
└─ ④ 都不是
└→ "chat"- 核心判定函数
looks_like_coding_prompt()用的是组合匹配策略：
输入: "帮我写一个 React 组件"
├─ 有文件名引用？(xxx.py, xxx.vue) → 没命中
├─ 有代码结构？(def, class, import) → 没命中
│
└─ 组合判定:
├─ 有动作词 "写" ✓
├─ 同时有产物对象 "组件" ✓
└─ 两个条件同时满足 → coding ✅
输入: "今天天气真好"
├─ 没文件名
├─ 没代码结构
├─ "天气" 不是动作词
└─ "真好" 不是技术对象 → chat ✅- 为什么必须组合匹配？ 单一关键词不可靠。"写"字可能出现在"写一封信"里。必须动作词 + 技术对象 同时出现才判为 coding。这是整个路由系统防误判的核心。但是后续我测试了几种情况，也会判断失误，而且会出现一些很奇怪的问题，不够我也改不动了
- 关键词检测函数：
| 函数 | 检测内容 | 示例命中 |
|---|---|---|
_has_coding_action() | 动作词 | 写、改、修、生成、创建、run、test、fix |
_has_workspace_object() | 工作对象 | 代码、文件、模块、组件、函数、bug |
_has_deliverable_object() | 产物对象 | demo、示例、页面、工具、插件 |
_has_tech_context() | 技术上下文 | python、vue、react、fastapi、git |
has_file_reference() | 文件名正则 | app.ts、utils.py |
_has_issue_keyword() | 错误关键词 | bug、报错、异常、traceback、failed |
工厂②：for_chat() —— chat 快速打包
def for_chat(cls, prompt, context):
return RoutingDecision(
intent="chat",
action_name="chat.reply", # 固定动作，不需要拆解
action_input={"prompt": prompt, "context": context},
reason="Chat intent detected.",
)- chat 是最简单的路径——不需要拆解具体动作，不需要提取参数。把原始 prompt 和上下文原样打包，L2 直接用自己的聊天人格和 LLM 对话生成回复。
分流③：_route_coding() —— 旧任务 vs 新请求
def _route_coding(self, prompt, context):
run_action = detect_run_action(prompt)
# 分支A: 用户在操作已有任务
if run_action != "create":
target_run_id = extract_run_reference(prompt)
return RoutingDecision.for_coding(
action_name={
"retry": "run.retry", # "重试一下 abc-123"
"rerun": "run.rerun", # "重新跑 abc-123"
"cancel": "run.cancel", # "取消 abc-123"
"inspect": "run.inspect", # "看看 abc-123 的进度"
}[run_action],
action_input={"run_id": target_run_id},
)
# 分支B: 全新的编程请求 → 进入动作分类器
action_name, action_input, reason = _classify_coding_action(prompt, context)
return RoutingDecision.for_coding(...)- 先判断是不是旧任务操作，如果是就跳过动作分类，直接走 run 控制。不是旧任务才进入四级分类器。
分类器④：_classify_coding_action() —— 四级分级
- 这是 L1 最核心的函数。把
coding意图拆解为具体动作 + 结构化参数。
_classify_coding_action(prompt, context)
│
│ 关键词匹配 → 命中哪层就进哪层，不往下走
│
├── Tier 1: 破坏性写操作 → 必须 LLM 提取参数
│ ├─ "写/生成/创建/实现" → workspace.write
│ ├─ "删除/remove/rm" → workspace.delete
│ ├─ "移动/重命名/move" → workspace.move
│ └─ "复制/拷贝/copy" → workspace.copy
│
├── Tier 2: 复杂只读操作 → LLM 提取参数
│ ├─ "测试/pytest/test" → workspace.test
│ └─ "搜索/查找/grep/find" → workspace.search
│
├── Tier 3: 简单读取 → 正则优先，失败才用 LLM
│ └─ "读/打开/查看/read" → workspace.read
│ ├─ 正则能提到路径 → 直接用（省一次 LLM 调用）
│ └─ 提不到 → LLM 提取
│
├── Tier 4: 零参数操作 → 直接执行
│ └─ "列出/目录/ls/tree" → workspace.list
│
└── Default: 兜底
└─ 都不命中 → run.create (当普通代码运行)- Tier 分级的逻辑：
| Tier | 动作类型 | 为什么用 LLM？ |
|---|---|---|
| 1 | write / delete / move / copy | 破坏性操作，必须精确拿到文件路径和完整内容 |
| 2 | test / search | 参数复杂（搜什么、哪个目录、哪个测试文件） |
| 3 | read | 大部分情况正则就能从 prompt 抓到路径，省一次 LLM 调用 |
| 4 | list | 完全不需要参数，跳过 LLM 直接执行 |
- LLM 参数提取——当 Tier 1/2 命中（或 Tier 3 正则失败）时触发：
用户: "帮我在 src/utils 下写一个 helper.py"
_extract_workspace_params_via_llm(prompt, context, "workspace.write")
│
├─ System Prompt: "你是参数提取器，只返回 JSON，不要废话"
├─ User Message: "Action: workspace.write
│ User request: 帮我在 src/utils 下写一个 helper.py"
├─ call_llm_sync() ← 轻量模型、低温、小 token 数
│
└─ LLM 返回:
{
"action": "workspace.write",
"rel_path": "src/utils/helper.py",
"content": "# Helper utilities\n\ndef ...",
"overwrite": false
}
LLM提参数：
- 意图判定（chat / coding）用关键词就够了，0ms，0 token
- 参数提取（"src/utils/helper.py"的精确路径、完整文件内容）关键词做不到，必须 LLM
- 因为关键词匹配可以判断出当前需要做的工作，但是无法确定除具体要写什么，这时候就要调用LLM提取出具体的参数。
call_llm_sync()本身是一个通用的 LLM 调用函数，用于参数提取只是他的使用场景之一，他在整个项目中被多次用到，但是每次的用途不一样，传入的system_prompt不一样，因此效果也不一样，LLM的行为不一样。
call_llm_sync(prompt="...", system_prompt=???)
兜底⑤：for_unknown()
def for_unknown(cls, prompt):
return RoutingDecision(
intent="unknown",
action_name="final.answer",
action_input={"content": "本机不太确定你想做什么呢...能再说清楚一点吗？"},
)- 触发条件：空消息、纯符号、乱码。让 L2 返回一句友好的引导语，不会静默失败。
产出：RoutingDecision
@dataclass(frozen=True, slots=True) # frozen = 不可变
class RoutingDecision:
intent: str # "chat" | "coding" | "unknown"
action_name: str # "chat.reply" | "workspace.write" | "run.create" | ...
action_input: dict # {"prompt":..., "rel_path":..., "content":..., ...}
reason: str # 为什么这么判（调试用）
metadata: dict # 额外信息（run_id 等）frozen=True 的意义：L1 判决完后，L2 和 L3 只能读不能改。防止下游代码意外污染路由决策，也防止并发问题。
设计原则
① 纯分类
→ 不执行任何工作，不调工具，只产出决策
→ 职责边界清晰，L1 绝不越界干活
② 关键词优先
→ 意图判定不用 LLM（快、免费、可预测、可调试）
→ 大模型留给真正需要语义理解的参数提取场景
③ LLM 最小化使用
→ 仅在关键词无法胜任的结构化参数提取时调用
→ 用了也是小模型 + 低温 + 小 token 数，把成本压到最低
④ 逐级降级
→ LLM 失败 → 正则兜底
→ 正则也失败 → run.create 最终兜底
→ 每一级都有 fallback，不会因为一个环节挂了就让整个请求失败总的来说
L1 是一个路由层。它用关键词做快速分类，用 LLM 做参数提取，用逐级降级保证鲁棒性。产出的 RoutingDecision 是不可变对象，下游只消费不管修改。整个系统的"理解用户意图"这件事，全压缩在这一层里。
Layer 2: Roleplay — 角色扮演层
L2接收 L1 的路由决策，调用 L3 干活，然后把L3返回的结果包装成 Shion 的口吻，推送到前端。
核心流程
process(decision, session_id, memory_context)
│
├─ chat 路径
│ ├─ _handle_chat()
│ │ ├─ 拿 decision.action_input 里的 prompt + context
│ │ └─ call_llm_sync() → CHAT_SYSTEM_PROMPT（纯聊天人格）
│ │
│ └─ 返回 ProcessResult（不调 L3）
│
└─ coding 路径
├─ work_agent.execute(decision) → L3 返回 work_result(dict)
├─ _generate_persona_response() → 把 dict 包装成 Shion 口吻
│ ├─ 失败 → 硬编码返回 sad 表情（不浪费 token）
│ └─ 成功 → 调 LLM，填 ROLEPLAY_SYSTEM_PROMPT + state_context
├─ mood 更新（连续成功→开心，连续失败→沮丧）
├─ _emit_to_frontend() → 推 4 路消息到前端
└─ 返回 ProcessResult(response, work_metadata)两个 System Prompt
CHAT_SYSTEM_PROMPT — 纯聊天用，简化版，没有状态占位符：
CHAT_SYSTEM_PROMPT = """
# 角色设定
你是"Shion"——一个寄宿在用户桌面上的 AI 智能体桌宠。
...（性格、语气、输出JSON格式要求）...
"""ROLEPLAY_SYSTEM_PROMPT— 工作结果包装用，带两个动态占位符：
ROLEPLAY_SYSTEM_PROMPT = """
# 角色设定
你是shion——...
当前状态上下文：
{state_context} ← 填入 L3 的工作结果摘要
当前情绪倾向：
{mood_modifier} ← 填入当前情绪描述
请生成回复。只输出 JSON！
"""- 调用前动态填入：
state_context = "意图: coding\n动作: workspace.write\n结果: 成功\n输出: 已创建 sort.py\n步数: 3"
mood_modifier = "你现在心情很好，充满能量。可以欢呼、用颜文字。"
system_prompt = ROLEPLAY_SYSTEM_PROMPT.format(
state_context=state_context,
mood_modifier=mood.modifier_text,
)- 两套 prompt 共享同一套角色人设，区别是聊天不需要任务状态，工作包装需要。
_generate_persona_response() — 角色包装
- 输入是 L3 返回的 dict：
work_result = {
"ok": True,
"summary": "已创建文件 src/sort.py",
"output": "def bubble_sort(arr): ...",
"metadata": {"run_id": "abc-123", "step_count": 3}
}- 处理逻辑：
① L3 失败？ → 不调 LLM，直接返回 RoleplayResponse(expression="sad", quip="唔...失败了")
② L3 成功 → 构建 state_context → 填 system_prompt → call_llm_sync()
→ LLM 返回 JSON: {"chat_line":"搞定啦！⭐","expression":"proud","quip":"机魂大悦！"}前端发送
_emit_to_frontend(response, ...)
├─ send_chat_message(chat_line) → 聊天窗口正文
├─ send_expression(expression) → Live2D 表情变化
├─ send_quip(quip) → 模型上方浮动气泡
└─ send_motion(motion) → Live2D 动作组- 四条消息异步并发推送。chat 路径用更简约的
_emit_chat_to_frontend()，只发消息和表情，不发气泡和动作。
情绪系统
全局单例
_session_mood = RoleplayMood()，跨请求累积：- 连续成功 ≥3 → happy → "你现在心情很好，充满能量。可以欢呼、用颜文字。"
- 连续失败 ≥3 → frustrated → "你现在有点烦躁，语气可以带刺，嘴硬心软。"
- 连续失败 1-2 → tired → "你现在有点累了...语气稍微慵懒，带点委屈。"
- 闲置 ≥5 轮 → lonely → "你现在感到寂寞...主动寻求关注，但不过度。"
- 其他 → neutral → "心态平稳，自然放松。"
每次 process() 结束时根据 ctx.scenario() 调 mood.record_success() / record_failure() / record_neutral()。下一轮请求的 mood_modifier 就会变，Shion 的语气就跟着变。
输出结构
ProcessResult(
response = RoleplayResponse(
chat_line, # 聊天文本
expression, # Live2D 表情名
quip, # 气泡短句
motion, # Live2D 动作组
),
work_metadata = {run_id, run_action, step_count, ...} # 给调度器用
)- 回到
services/chat.py：chat_line→ HTTP 响应，work_metadata→ 调度 + 持久化。
Layer 3: WorkAgent — workflow
L3接收 L2 传来的路由决策，执行具体的工作，返回结构化结果。不做任何角色扮演，不推前端消息。
在三层中的位置
L2: roleplay_agent.process()
│
├─ chat 意图 → L2 自己处理，不调 L3
│
└─ coding 意图 → work_agent.execute(decision) ← L3 入口
│
├─ 普通文件操作 (write/read/list/delete...)
│ → action registry 直接执行
│
└─ run.create（生成+运行代码）
→ LangGraph 子图：PM → Coder → Executor → QA → Debuggerwork_agent.execute() — 入口
def execute(self, decision, *, session_id, turn_id, memory_context):
if decision.intent == "chat":
return self._chat_result(decision) # {"ok": True, "summary": ""}
if decision.intent == "coding":
return self._execute_loop(decision, ...) # LangGraph 循环
return {"ok": False, "error": "Unknown intent"}- chat 意图不经过 L3，L2 已经从
process()里截断了。coding 意图进_execute_loop()。
_execute_loop() — LangGraph 循环
def _execute_loop(self, decision, *, session_id, turn_id, memory_context):
# 拼 prompt + context
prompt = decision.action_input.get("prompt")
context = decision.action_input.get("context")
if memory_context:
context = f"{memory_context}\n\n---\n\n{context}"
# 启动 LangGraph
result = run_agent_loop(
prompt=prompt,
context=context,
action_name=decision.action_name, # "workspace.write"
action_input=decision.action_input, # {rel_path, content, ...}
session_id=session_id,
)
# 从 state 提取结果
return {
"ok": result.state["action_result"]["ok"],
"summary": result.state["output"],
"output": result.state["output"],
"metadata": {run_id, run_action, step_count, stop_reason, ...}
}LangGraph 图结构
┌──────────────┐
│ plan_node │ ← 入口：把 L1 的路由决策转成执行计划
└──────┬───────┘
│
▼
┌──────────────┐
│ act_node │ ← 核心：执行动作
└──────┬───────┘
│
▼
┌──────────────┐
│ observe_node │ ← 观察结果，标记 done=True
└──────┬───────┘
│
▼
┌──────────────────┐
│decide_continue │ ← 判断：结束了还是继续循环？
└──┬───────────┬───┘
│ │
done=True│ │done=False
▼ ▼
┌──────────┐ ┌──────────┐
│finalize │ │ plan_node │ ← 重新计划（目前一次就停）
└──────────┘ └──────────┘- 现在是单轮循环——
observe_node里直接done = True，执行完一个动作就结束了。
act_node — 动作分发
def _execute_selected_action(state):
action_name = state["action_name"]
# 分支1: run.create → 进入 coding 子图（PM→Coder→Executor→QA→Debugger）
if action_name in {"run.create"}:
return _execute_action_via_coding_workflow(state)
# 分支2: 普通文件操作 → 走 action registry
return default_action_registry.execute(action_name, state["action_input"])| action_name | 走向 | 做什么 |
|---|---|---|
workspace.write | action registry | 把 content 写入文件 |
workspace.read | action registry | 读取文件内容并返回 |
workspace.list | action registry | 列出目录结构 |
workspace.delete | action registry | 删除文件 |
workspace.search | action registry | 搜索文件内容 |
run.create | coding 子图 | 生成代码 → 写入 → 执行 → 测试 → 修复 |
Coding 子图 --run.create
- 用户说"帮我写个 Python 脚本"时，L1 判定为
run.create，L3 进入这个子图：
┌──────────┐
│ pm_node │ PM（项目经理）：调 LLM，把需求拆成编码任务
└────┬─────┘
│
▼
┌──────────┐
│coder_node│ Coder（程序员）：调 LLM，生成代码
└────┬─────┘
│
▼
┌──────────────┐
│executor_node │ Executor（执行器）：把代码写入文件并运行
└────┬─────────┘
│
▼
┌──────────┐
│ qa_node │ QA（测试）：检查执行结果，判断是否通过
└────┬─────┘
│
┌────┴────┐
│ │
通过? 失败?
│ │
▼ ▼
┌─────────┐ ┌──────────────┐
│ finish │ │ debugger_node│ Debugger（调试员）：调 LLM 分析错误并修复
└─────────┘ └──────┬───────┘
│
▼
┌──────────┐
│coder_node│ ← 回到 Coder 重新生成修复后的代码
└──────────┘- 这五个角色全是同一套 LLM + 不同的 system prompt。简单来说就是PM 做任务拆解，Coder 写代码，Executor 运行，QA 检查结果，Debugger 分析错误。
engine.py 中的 WORK_SYSTEM_PROMPT
WORK_SYSTEM_PROMPT = """你是 Layer 3 工作引擎。
只负责纯粹的代码执行和文件操作，不承担任何角色扮演。
核心规则:
1. 专注任务：只关注用户的实际工作需求，忽略闲聊
2. 文件操作：所有读写都在安全的 workspace 内进行
3. 结果导向：执行完成后给出结构化的结果输出
4. 错误透明：遇到错误时如实报告，不含糊
输出格式: {"ok": true|false, "summary": "...", "output": "...", "error": "..."}
"""- 这个 prompt 让 L3 只说工作数据，返回
{"ok": true, "summary": "文件已创建"}这纯技术输出。
L3 返回给 L2 的结果
{
"ok": True, # 执行是否成功
"intent": "coding",
"summary": "已创建文件 src/sort.py", # 人类可读摘要
"action_name": "workspace.write",
"output": "def bubble_sort(arr):...", # 完整输出
"error": None,
"file_context": {"recent_files": [...], "last_modified": "..."},
"metadata": {
"run_id": "abc-123",
"run_action": "create",
"step_count": 3,
"stop_reason": "completed"
}
}- 这个 dict 回到 L2 后，L2 用
_generate_persona_response()把它包装成 Shion 的口吻。
三层之间的通信协议
- L1、L2、L3 在同一进程内运行，之间传 Python 对象，没有序列化开销。每一层只暴露一个入口函数，输入和输出都有明确的类型。
通信拓扑
L1: RoutingGuard
route(prompt, context, file_context)
│
│ RoutingDecision (frozen dataclass)
▼
L2: RoleplayAgent
process(decision, session_id, memory_context)
│ │
│ chat → 自己处理 │ coding → 调 L3
│ │
│ │ RoutingDecision (同一个对象)
│ ▼
│ L3: WorkAgent
│ execute(decision, session_id, ...)
│ │
│ │ dict {ok, summary, output, metadata}
│ ▼
│ L2 拿回结果 → 角色包装 → 前端发射
│ │
◄────────────────────┘
│
│ ProcessResult (response + work_metadata)
▼
services/chat.py① L1 → L2：RoutingDecision
- 谁调谁：
services/chat.py调 L1，拿到RoutingDecision，然后传给 L2。
# services/chat.py 第216-231行
decision = routing_guard.route(prompt, effective_context, file_context)
# ↑ L1 的入口函数
process_result = await anyio.to_thread.run_sync(
partial(roleplay_agent.process, decision, ...)
# ↑ 同一个对象传给 L2
)- 传的是什么：
@dataclass(frozen=True) # 不可变，L2/L3 只能读不能改
class RoutingDecision:
intent: str # "chat" | "coding" | "unknown"
action_name: str # "chat.reply" | "workspace.write" | "run.create" | ...
action_input: dict # {prompt, context, rel_path, content, ...}
reason: str # 调试用
metadata: dict # 额外信息- frozen：防止 L2 或 L3 意外修改决策。L1 判完就封存。
② L2 → L3：同一个 RoutingDecision
- L2 的
process()方法调 L3 的execute()。
# roleplay.py 第621-626行
work_result = self.work_agent.execute(
decision, # ← L1 产出的同一个 RoutingDecision 对象，原样传入
session_id=session_id,
turn_id=turn_id,
memory_context=memory_context,
)- L2 不对
decision做任何修改，直接转发。L2 的角色只是"中间人"——聊天时自己处理，coding 时转给 L3。
③ L3 → L2：dict
- L3 的
execute()返回一个 dict 给 L2。
# engine.py WorkAgent._execute_loop() 返回
return {
"ok": True,
"intent": "coding",
"summary": "已创建文件 src/sort.py",
"action_name": "workspace.write",
"output": "def bubble_sort(arr): ...",
"error": None,
"file_context": {"recent_files": ["src/sort.py"]},
"metadata": {
"run_id": "abc-123",
"run_action": "create",
"step_count": 3,
"stop_reason": "completed",
},
}- 为什么用 dict 而不是 dataclass：L3 是纯工作引擎，不应该依赖 L2 的数据结构。dict 是通用契约，L2 自己知道怎么解析。
L2 拿到后：
- 从
work_result提取字段 → 拼state_context→ 调 LLM 做角色包装 - 从
work_result["metadata"]提取run_id/run_action→ 放入ProcessResult.work_metadata
- 从
④ L2 → services：ProcessResult
# roleplay.py
@dataclass
class ProcessResult:
response: RoleplayResponse # 角色包装后的内容
work_metadata: dict[str, Any] # L3 的元数据
@dataclass
class RoleplayResponse:
chat_line: str # "搞定啦！已创建 sort.py ⭐"
expression: str # "proud"
quip: str # "机魂大悦！"
motion: str # ""
mood_label: str # "happy"
scenario: str # "success"- 回到
services/chat.py后：
# response.chat_line → HTTP 响应正文
# work_metadata → 调度后台执行 + 持久化
result = ChatServiceResult(
output=process_result.response.chat_line,
run_id=process_result.work_metadata.get("run_id"),
...
)数据流汇总
L1 产出 L2 消费 L3 消费
─────────────────────────────────────────────────
RoutingDecision ──→ RoutingDecision ──→ RoutingDecision
(chat:自己用) (coding:执行)
L3 产出
────────
dict {ok, summary, output, metadata}
│
▼
L2 消费 → 角色包装
│
▼
ProcessResult(response, work_metadata)
│
▼
services/chat.py
├─ response.chat_line → HTTP
└─ work_metadata → 调度+持久化设计要点
① 同进程传对象 → Python 对象直接传，不走网络，不走 JSON
② 单向依赖 → L1 → L2 → L3，后层不调前层
③ frozen 防篡改 → RoutingDecision 创建后不可改
④ dict 解耦 → L3 不依赖 L2 的数据结构
⑤ L2 是唯一枢纽 → 所有路径都经过 L2，它是三层的"调度中心"
评论已关闭