Helix 智能体架构

Helix 核心接口、消息流转与后端架构

本文按当前代码（services/app + apps/web）描述：HTTP 接口、发送消息后的处理链路、SSE 的完整用法，以及后端每个类的职责。实现以源码为准，路径相对于仓库根目录。

1. 核心接口

所有 HTTP 入口集中在 HelixController，前缀 /api/v1。前端 apps/web/src/api.ts 的 REST 走 Vite 代理 /api → localhost:8080；SSE 在开发环境直连 http://localhost:8080，避免代理把流攒批。

1.1 接口一览

可选鉴权：配置了 HELIX_API_TOKEN 时，ApiTokenFilter 要求 X-API-Token 或 Authorization: Bearer。/health 与 /actuator/health 放行。

1.2 三种推理模式

POST .../messages 的 mode：

2. 发送消息后的处理与流转

这是系统主路径。前端 send() → REST 落库 → 后台线程跑 Agent → SSE 推增量 → 结束后前端拉全量消息。

2.1 总览

sequenceDiagram
    participant UI as apps/web
    participant API as HelixController
    participant CS as ConversationService
    participant Loop as AgentLoop
    participant Hub as EventHub
    participant GW as ModelGateway
    participant NIM as NVIDIA /v1/chat/completions

    UI->>API: POST /conversations/{id}/messages
    API->>CS: postMessage(text, modelOverride, mode)
    CS->>CS: 写 user Message + Run(CREATED)
    CS->>Loop: start(runId)  // helixExecutor 异步
    CS-->>UI: { message, run: { id, state: CREATED } }
    UI->>API: GET /runs/{id}/events (fetch SSE)
    API->>Hub: subscribe(runId, after)
    Hub-->>UI: heartbeat + 历史非 token 事件
    Loop->>Loop: CREATED → EXECUTING
    Loop->>Hub: status thinking
    Loop->>GW: chat(window, tools, onToken)
    GW->>NIM: stream=true
    NIM-->>GW: delta.reasoning / content
    GW-->>Loop: onToken(piece)
    Loop->>Hub: token
    Hub-->>UI: event: token
    alt 模型返回 tool_calls
        Loop->>Hub: tool.start
        Loop->>Loop: 执行工具或 AWAIT_APPROVAL
        Loop->>Loop: continue 下一轮
    else 纯文本
        Loop->>Loop: 写 assistant Message
        Loop->>Hub: done SUCCEEDED
        Hub-->>UI: event: done（然后 complete emitter）
        UI->>API: GET messages + GET run
    end

2.2 前端发送（App.tsx → api.postMessage）

1. 可选先 POST /conversations（无会话时）。
2. 乐观插入一条本地 user 气泡。
3. POST /api/v1/conversations/{id}/messages，body：

{ "text": "...", "mode": "react", "modelOverride": { "provider": "nvidia", "modelId": "openai/gpt-oss-120b" } }

modelOverride 仅在用户改过下拉框时带上。

4. 用返回的 run.id 调用 subscribeRun，同时 4s 轮询 GET /runs/{id} 作兜底（SSE 丢了也能在终态拉消息）。

2.3 ConversationService.postMessage

实现：services/app/.../conversation/ConversationService.java。

1. 校验 会话存在、text 非空。
2. 选模型 ModelRouter.resolve(override, conversation.model, workspace.settings)
   优先级：请求覆盖 > 会话已存模型 > 工作空间默认 > HelixProperties（默认 nvidia / minimaxai/minimax-m3）。
3. 写 user 消息 role=user，id 形如 msg_...。
4. 写 Run：state=CREATED，goal=text，mode，modelRefJson，maxSteps / 日预算来自 workspace 或环境。
5. 把 message.runId 指到该 run。
6. 启动：默认 helix.sync-runs=false，调用 AgentLoop.start（线程池 helix-run-*）。测试 profile 可同步 execute。
7. 立即返回 { message, run: { id, state } }。此时模型往往还没开始，state 仍是 CREATED。真正的助手内容不走这个响应，走 SSE + 事后 GET messages。

2.4 AgentLoop.execute：状态机入口

合法迁移见 RunFsm：

CREATED → PLANNING | EXECUTING | FAILED | CANCELLED | BUDGET_EXCEEDED
PLANNING → AWAIT_APPROVAL | EXECUTING | ...
AWAIT_APPROVAL → EXECUTING | CANCELLED | FAILED
EXECUTING → AWAIT_APPROVAL | SUCCEEDED | FAILED | CANCELLED | BUDGET_EXCEEDED | AWAIT_USER

CREATED 时：

• mode=plan 且还没有 plan → PLANNING，planOnce（不带 tools），解析 JSON 计划，发 plan 事件，转入 AWAIT_APPROVAL，本轮返回，等 POST /runs/{id}/resume。
• 否则 → EXECUTING，进入 loop。

异常：describeFailure 把 429 / 超时 / 404 Function / 错误 model id / 鉴权映射成中文，写入 assistant 消息 + run.error，发 error 再 done。

2.5 loop：一轮或多轮模型调用

每一轮：

1. 取消 / 日预算 / maxSteps 检查。
2. bumpStep + checkpoint。
3. 拼上下文 MemoryService.window：
   • system = SystemPrompt（工具列表 + 引用规范）
   • 超过 24 条会先让模型做摘要（SUMMARY_PLEASE）
   • 再带最近 16 条（含 tool 角色）
4. direct 则 tools=[]，否则 ToolHub.listSpecs()。
5. gateway.chat(..., onToken)：
   • NVIDIA：CompletionsStream HTTP stream=true，把 delta.content 以及 gpt-oss 的 reasoning / reasoning_content 回调出来。
   • 每段 onToken → EventHub.emit("token", {text})。
   • 流失败且非 404/429/超时 → 回退非流式 OpenAiChatModel；若此时 streamed 仍空，整段再发一次 token。
6. 有 tool_calls：
   • 落一条 assistant 消息（parts.toolCalls）
   • 每个 call：tool.start → 高风险且未批准则 approval + AWAIT_APPROVAL 并 return
   • 否则 ToolHub.invoke，观察写入 role=tool 消息，tool.end，continue 下一轮（模型能看见 tool 结果）
7. 无 tool_calls：落 assistant 文本，SUCCEEDED，done。

高风险工具（当前 code_exec）审批通过后：decide → EXECUTING → finishPendingTool 真正执行，再继续 loop。

2.6 消息在库里的形态

表 messages：

前端在 done 后 GET /messages 用库里的完整消息替换实时气泡（liveText 清空）。因此 持久化真相是消息表，SSE 只负责过程。

2.7 模型调用原理（NVIDIA）

AgentLoop
  → ModelGateway.clientFor(ModelRef)
       无 Key / HELIX_FAKE_MODEL → FakeModel
       nvidia → OpenAiCompatProvider(base=https://integrate.api.nvidia.com/v1)
  → CompletionsStream.execute
       POST {base}/chat/completions  stream=true
       解析 SSE data: 行
       content 优先作为最终 ChatResult.content
       content 为空时用 reasoning 填可见流

模型 ID：ModelCatalogService.canonicalModelId 把 gpt-oss-120b 映射成 openai/gpt-oss-120b。配置表 model_configs 提供 temperature / top_p / max_tokens / tools 开关。

3. SSE 在本项目中的用法

Helix 不用 把 LLM 的 HTTP 流直接转给浏览器。而是：

1. 后端自己连 NVIDIA（流式）。
2. 把增量翻译成 按 run 隔离的应用层事件。
3. 浏览器订 GET /runs/{runId}/events。

NVIDIA 的 text/event-stream 与浏览器的 SSE 是两段独立连接。

3.1 为什么要自己做一层

• 一次 run 里不只有 token：还有 status、工具、审批、计划、引用、结束。
• 刷新 / 切会话后要 续订，不能只靠 TCP 不断。
• Vite 代理会缓冲 chunked 响应；开发态 SSE 必须直连 8080。
• gpt-oss 的 token 在 reasoning_content 里，必须在服务端拆包后再转成统一的 token 事件。

3.2 服务端：EventHub + SseEmitter

文件：runtime/EventHub.java。按 runId 维护三张并发表：

订阅 subscribe(runId, lastEventId)：

1. new SseEmitter(10 分钟)。
2. 注册 onCompletion / onTimeout / onError → 从 subs 摘掉。
3. 立刻发 event: heartbeat（让响应提交、代理别缓冲）。
4. 回放 history 里 id > cursor 的事件；碰到 done 则 complete()。
5. 返回 emitter，Spring MVC 保持异步请求。

Cursor：?after= 优先，否则请求头 Last-Event-ID。

发送 emit(runId, event, data)：

1. id = seq++。
2. event != "token" 才写入 history（避免重连时把几百个 token 一次性倒回）。
3. 向该 run 所有 emitter send(id + name + JSON data)。
4. 发送失败且不是客户端断开 → complete()；客户端断开则只 drop。
5. done 后 complete 并 drop。

心跳：@Scheduled(fixedRate = 8000) 对所有订阅发 heartbeat。失败的连接被清掉。这不是给 UI 用的，是为了保活和探测死连接。

控制器在订阅前设置：

• Content-Type: text/event-stream;charset=UTF-8
• Cache-Control: no-cache, no-transform
• X-Accel-Buffering: no
• Connection: keep-alive
• 若有 Origin，回写 Access-Control-Allow-Origin（开发态 5173 → 8080）

线上 Tomcat 异步请求超时：spring.mvc.async.request-timeout: 600000。客户端中止连接由 ApiExceptionHandler 吞掉，避免堆栈刷屏。

3.3 事件目录

浏览器只处理 命名事件（event: 字段）。heartbeat 丢弃。

SSE 线上形态示例：

id: 0
event: heartbeat
data: {"ok":true}

id: 1
event: status
data: {"phase":"thinking","text":"正在思考"}

id: 2
event: token
data: {"text":"The"}

id: 15
event: done
data: {"runId":"run_abc","state":"SUCCEEDED"}

3.4 客户端：subscribeRun（api.ts）

不用 EventSource。原生 EventSource 在 CONNECTING 时也会打 error，旧逻辑会关连接再重连，把积压 token 一次画完。

现行方案：fetch + ReadableStream + AbortController。

sseOrigin() =
  VITE_API_BASE 或（DEV）http://localhost:8080
GET {origin}/api/v1/runs/{runId}/events?after={lastId}&token=
Accept: text/event-stream
可选 X-API-Token

读循环：

1. TextDecoder 追加到 buffer，按 \n\n 切 SSE 块。
2. dispatchSseBlock 解析 id / event / data: 多行。
3. token → onToken；done/error → finished=true，reader.cancel()。
4. 网络失败且未 abort：指数退避重连，带上已记录的 lastId（after）。
   token 不在 history 里，重连不会倒字；最多丢未持久化的实时字，终态靠 GET messages 补齐。
5. close()：stopped=true + abort()。

App.tsx 的 attachStream：同一 runId 已有流则 resume 直接返回，避免切会话 effect 把活连接掐掉。done 后拉全量消息，用库数据替换 liveText。

3.5 与 NVIDIA 流的对照

两段流失败互不影响：NVIDIA 断了由 AgentLoop 记 FAILED；浏览器断了只丢实时通道，run 继续。

3.6 续订与刷新

用户刷新或从侧栏点回正在跑的会话：

1. GET /messages 画已落库部分。
2. GET .../runs/latest，若 state 仍在 LIVE（CREATED/PLANNING/EXECUTING/AWAIT_*），attachStream(runId, { resume: true })。
3. SSE 回放 history 中的 status/plan/tool/approval（token 不回放）。
4. 新 token 继续追加；若 run 已 done，回放里带 done，前端再拉一次 messages。

4. 后端代码架构与类职责

HelixApplication          启动、线程池、种子 workspace/KB/模型表
HelixController           唯一 REST + SSE 适配器
ApiExceptionHandler       HTTP 错误码、客户端中止静默
        │
        ├─ conversation/  会话与消息持久化
        ├─ runtime/       Run 状态机、Agent 循环、SSE 总线、审批
        ├─ model/         模型目录、路由、NVIDIA/OpenAI/Fake
        ├─ memory/        滑动窗口与摘要
        ├─ tool/          工具注册与执行
        ├─ rag/           知识库切片检索
        ├─ observe/       轨迹与日预算
        └─ security/      CORS、API Token

包结构对应 com.helix.*。运行时一个 JVM 进程，H2（local）或 Postgres；模型 按请求构建，没有进程级 ChatModel 单例。

4.1 启动与配置

4.2 API 层

4.3 会话域 conversation

4.4 运行时 runtime

4.5 模型 model

4.6 记忆 memory

4.7 工具 tool

4.8 RAG rag

4.9 可观测 observe

4.10 安全 security

5. 一张图看完主链路上的对象

用户输入
  → ConversationService.postMessage
       Message(user) + Run(CREATED)
  → AgentLoop (helix-run 线程)
       MemoryService.window → 消息列表
       ModelGateway → CompletionsStream → NVIDIA
       EventHub.emit(token/status/tool/...)
       ToolHub.invoke → Message(tool)
       Message(assistant) + Run(SUCCEEDED|FAILED|AWAIT_APPROVAL|...)
  → GET /runs/{id}/events  (EventHub / SseEmitter)
  → 浏览器 fetch 流 → liveText
  → done → GET /messages 替换为库中记录

相关源码入口：

• 接口：services/app/src/main/java/com/helix/api/HelixController.java
• 发消息：.../conversation/ConversationService.java
• 循环：.../runtime/AgentLoop.java
• SSE：.../runtime/EventHub.java + apps/web/src/api.ts（subscribeRun）
• 模型流：.../model/CompletionsStream.java