mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-22 18:08:04 +00:00
895948b24e
实测：停止后排队消息仍逐条跑完。原因是 armStop 只数 sched.queue， 而用户按下停止时调度器正忙于当前任务，其余消息大多还没被 pumpInbox 搬进队列、仍停在 inputCh ⇒ queued=0、配额归零。 - IOManager.PendingInputs()：暴露 channel 中待处理条数。 - armStop(pending int)：queued = len(s.queue) + pending。 - 补 TestStop_ArmCountsPendingChannelInputs 锁死该口径。
485 lines
18 KiB
Go
485 lines
18 KiB
Go
package core
|
||
|
||
import (
|
||
"fmt"
|
||
"log"
|
||
"runtime/debug"
|
||
"time"
|
||
|
||
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
|
||
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
|
||
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||
)
|
||
|
||
// eventLoop 已由 scheduler.go 的 schedulerLoop 取代（M2）。
|
||
//
|
||
// 原实现直接在 select 里处理 inputCh/selfInputCh，没有任何可枚举的队列、
|
||
// 无法承载优先级与抢占；现在任务先入就绪队列，由选择函数 pickTaskIndex 决定下一个。
|
||
// 兼容性说明：M2 全部任务为 LevelBackground，因此行为等价于原先的 FIFO。
|
||
|
||
func (a *Agent) interceptLoop() {
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
log.Printf("[agent] interceptLoop panic recovered: %v\n%s", r, debug.Stack())
|
||
time.Sleep(time.Second)
|
||
go a.interceptLoop()
|
||
}
|
||
}()
|
||
for {
|
||
select {
|
||
case evt := <-a.io.InputInterruptChan():
|
||
text, _ := evt.Payload["content"].(string)
|
||
stop, _ := evt.Payload["stop"].(bool)
|
||
if text == "" && !stop {
|
||
// 没有内容也不是停止指令：没有可处理的东西（旧行为）。
|
||
//
|
||
// 注意：**不能**把“空内容”一律当成空操作。客户端停止按钮
|
||
// 本来就不带消息（/chat/interrupt 收 body 空的 {}），
|
||
// 旧代码在这里 continue 掉，于是停止按钮毫无反应，
|
||
// 而且接口还回 200 骗调用方——已实测：HTTP 200 但内核零日志、
|
||
// 生成继续跑到自然结束。
|
||
continue
|
||
}
|
||
if stop {
|
||
// 停止：①立即结束当前 LLM 推理；②登记短路配额。
|
||
//
|
||
// 注意这里**只 arm、不 take**：takeStop 必须由 stepLLM 去消费，
|
||
// 它才是决定“取消后不重跑”的那个人。曾经写成
|
||
//
|
||
// if n := armStop(); n > 0 || takeStop() { ... }
|
||
//
|
||
// 这个 `||` 在 queued=0 时会短路到 takeStop()，把标记先消费掉，
|
||
// 于是 stepLLM 永远看不到它 → 取消后照样重跑一轮。
|
||
// 实测：停止被正确记录（`stop requested ... queued=0`）但生成仍跑到自然结束。
|
||
// pending 必须算上**停在输入 channel 里**的那一段：停止时
|
||
// 调度器多在半路忙当前任务，其余消息还没被 pumpInbox 搬进队列，
|
||
// 只数 sched.queue 会得到 0，配额随之失效（实测过）。
|
||
pending := 0
|
||
if a.io != nil {
|
||
pending = a.io.PendingInputs()
|
||
}
|
||
n := a.sched.armStop(pending)
|
||
log.Printf("[agent] stop requested by %s/%s (queued=%d will be short-circuited at pre-action)",
|
||
evt.Source, evt.OutputChannel, n)
|
||
a.cancelCurrentLLM()
|
||
if text == "" {
|
||
// 纯停止：不进中断队列、不产生新任务。旧实现把空停止当成一条
|
||
// 中断入队，取消后会以空内容重跑一轮，停下之后又“活着”。
|
||
continue
|
||
}
|
||
// 带注释的停止（/stop 说句话）：注释本身仍作为中断处理，
|
||
// 走下面的正常路径——用户想看模型对被停下话题的回应。
|
||
}
|
||
log.Printf("[agent] interrupt from %s/%s: %s", evt.Source, evt.OutputChannel, truncateStr(text, 80))
|
||
|
||
clone := &agentIO.InputEvent{
|
||
RequestID: evt.RequestID,
|
||
Source: evt.Source,
|
||
Type: evt.Type,
|
||
Payload: map[string]interface{}{},
|
||
OutputChannel: evt.OutputChannel,
|
||
}
|
||
for k, v := range evt.Payload {
|
||
clone.Payload[k] = v
|
||
}
|
||
clone.Payload["interrupt"] = true
|
||
clone.Payload["interrupt_source"] = evt.Source
|
||
clone.Payload["interrupt_channel"] = evt.OutputChannel
|
||
|
||
// 决策交给调度器：requestPreempt 总会登记中断（进中断队列或 immediate，
|
||
// 因而不会丢），仅当它会真抢占时才告诉我“该取消可取消的步骤”。
|
||
// 本 goroutine 不碰任何帧——只写中断队列与让位信号。
|
||
//
|
||
// 级别由来源声明（InjectOptions.Priority → payload["priority"]）；
|
||
// 未声明一律 L1。L4（“立即打断”）只有内核级插件能声明，
|
||
// 外部插件即便报了 L4 也会被夹到 L3；内核自身另有 raiseKernelInterrupt。
|
||
level := interruptLevel(evt, a.isKernelLevelSource(evt.Source))
|
||
if a.sched.requestPreempt(clone, level) {
|
||
a.cancelCurrentLLM()
|
||
}
|
||
|
||
case <-a.ctx.Done():
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// cancelCurrentLLM 取消正在进行的 LLM 请求（若有）。
|
||
//
|
||
// 只有 LLM 流式步骤是可取消的；工具 RPC / ONNX / CAS 在 v1 是临界区，
|
||
// 取消对它们无效——让位信号会等它们自然结束后的安全点（设计文档 D2）。
|
||
func (a *Agent) cancelCurrentLLM() {
|
||
a.llmMu.Lock()
|
||
cancel := a.cancelLLM
|
||
a.llmMu.Unlock()
|
||
if cancel != nil {
|
||
cancel()
|
||
log.Printf("[agent] LLM request cancelled by preemption")
|
||
}
|
||
}
|
||
|
||
// channelConsolidation 标记记忆整理类自输入：无记忆路径处理，
|
||
// 不写入对话上下文、不向任何输出通道 emit 响应。
|
||
const channelConsolidation = "_consolidation_"
|
||
|
||
// selfInputMsg 自循环输入消息。channel 决定处理路径：
|
||
// - channelConsolidation：记忆整理，无记忆（不污染上下文/知识库）
|
||
// - 其他值（如 "cli"、"webui"）：正常输入路径，写入上下文并 emit 响应
|
||
// （典型场景：子 Agent 完成通知，需让父 Agent 感知并可回复用户）
|
||
type selfInputMsg struct {
|
||
text string
|
||
channel string
|
||
}
|
||
|
||
// selfEvent 把内核自循环消息归一成输入事件。
|
||
func selfEvent(msg selfInputMsg) *agentIO.InputEvent {
|
||
if msg.channel == "" {
|
||
msg.channel = channelConsolidation // 兼容空值：默认走整理路径
|
||
}
|
||
return &agentIO.InputEvent{
|
||
Source: "system",
|
||
Type: "text",
|
||
Payload: map[string]interface{}{"content": msg.text},
|
||
OutputChannel: msg.channel,
|
||
}
|
||
}
|
||
|
||
func (a *Agent) handleSelfInput(msg selfInputMsg) {
|
||
_, _ = a.runInputTask(selfEvent(msg))
|
||
}
|
||
|
||
func (a *Agent) handleInput(evt *agentIO.InputEvent) {
|
||
switch evt.Type {
|
||
case "text", "image", "audio":
|
||
_, _ = a.runInputTask(evt)
|
||
|
||
case "event":
|
||
log.Printf("[agent] event from %s: %v", evt.Source, evt.Payload)
|
||
|
||
case "command":
|
||
cmd, _ := evt.Payload["command"].(string)
|
||
log.Printf("[agent] command from %s: %s", evt.Source, cmd)
|
||
|
||
default:
|
||
log.Printf("[agent] unknown event type from %s: %s", evt.Source, evt.Type)
|
||
}
|
||
}
|
||
|
||
// inputPayload 是一次输入在「模态」这个维度上的全部内容。
|
||
//
|
||
// 拆出这个结构，是为了让 processInput 只有一条主干：模态不再决定走哪个函数，
|
||
// 只决定这里的字段填不填。此前 text 与 image/audio 各有一个 process 函数，
|
||
// 媒体那条缺了去重、no_memory、通道 Cleaner、中断语义、EventRawInput 五项——
|
||
// 不是因为媒体不需要，而是复制粘贴之后文本那条继续演进、媒体那条没跟上。
|
||
type inputPayload struct {
|
||
// text 是进 LLM 与记忆的文本。纯媒体输入时它是 mediaToBlocks 给的 alt 文案。
|
||
text string
|
||
// blocks 非空表示本轮带多模态内容，随当前轮的 message 一起发给模型。
|
||
blocks []agentAPI.ContentBlock
|
||
// mediaType 供插件在 stage 里判断本轮媒体的模态。
|
||
mediaType string
|
||
// captureTool 是媒体落进 CAS 时记录的来源标签。
|
||
captureTool string
|
||
}
|
||
|
||
// resolveInput 把 InputEvent 归一成 inputPayload。
|
||
//
|
||
// 三种来源在这里合流：
|
||
// 1. evt.Type 是 image/audio —— 用户直接发的媒体，payload 里是 data/url；
|
||
// 2. evt.Type 是 text 且 payload 带 media_blocks —— 插件经 IOInjector 的
|
||
// InjectInputMedia / InjectInputMediaSync / InjectInterruptMedia 注入的
|
||
// 媒体，块已经是成品；
|
||
// 3. 纯文本。
|
||
//
|
||
// 第 2 种此前无处可去：注入方把块放进 payload，而文本路径不看这个键，
|
||
// 于是插件注入的媒体到 payload 就断了，且不报错。
|
||
func (a *Agent) resolveInput(evt *agentIO.InputEvent) (inputPayload, bool) {
|
||
switch evt.Type {
|
||
case "image", "audio":
|
||
blocks, alt := a.mediaToBlocks(evt.Payload, evt.Type, evt.Source)
|
||
return inputPayload{
|
||
text: alt,
|
||
blocks: blocks,
|
||
mediaType: evt.Type,
|
||
captureTool: "input_" + evt.Type,
|
||
}, true
|
||
}
|
||
|
||
text, _ := evt.Payload["content"].(string)
|
||
blocks, mediaType := injectedBlocks(evt.Payload)
|
||
// 文本与媒体都空才算无效输入：只带图不带字是合法的（插件注入常这样）。
|
||
if text == "" && len(blocks) == 0 {
|
||
return inputPayload{}, false
|
||
}
|
||
return inputPayload{
|
||
text: text,
|
||
blocks: blocks,
|
||
mediaType: mediaType,
|
||
captureTool: "inject_" + evt.Source,
|
||
}, true
|
||
}
|
||
|
||
// injectedBlocks 取出 payload 里插件注入的多模态块。
|
||
//
|
||
// 两种静态类型都要认：内核内部注入直接给 []agentAPI.ContentBlock，
|
||
// 而经公共 SDK 的 IOInjector 过来的是 []pubsdk.ContentBlock。两者字段完全一致，
|
||
// 但 Go 不会自动转换，只认一种的后果是另一种被静默丢弃。
|
||
func injectedBlocks(payload map[string]interface{}) ([]agentAPI.ContentBlock, string) {
|
||
var blocks []agentAPI.ContentBlock
|
||
switch v := payload["media_blocks"].(type) {
|
||
case []agentAPI.ContentBlock:
|
||
blocks = v
|
||
case []pubsdk.ContentBlock:
|
||
blocks = make([]agentAPI.ContentBlock, 0, len(v))
|
||
for _, b := range v {
|
||
nb := agentAPI.ContentBlock{Type: b.Type, Text: b.Text}
|
||
if b.ImageURL != nil {
|
||
nb.ImageURL = &agentAPI.ImageURL{URL: b.ImageURL.URL, Detail: b.ImageURL.Detail}
|
||
}
|
||
if b.AudioURL != nil {
|
||
nb.AudioURL = &agentAPI.AudioURL{URL: b.AudioURL.URL}
|
||
}
|
||
blocks = append(blocks, nb)
|
||
}
|
||
}
|
||
if len(blocks) == 0 {
|
||
return nil, ""
|
||
}
|
||
// 模态由块自身判定，注入方不必额外声明。图优先：一次注入里图片是主体。
|
||
mediaType := ""
|
||
for _, b := range blocks {
|
||
if b.ImageURL != nil {
|
||
return blocks, "image"
|
||
}
|
||
if b.AudioURL != nil {
|
||
mediaType = "audio"
|
||
}
|
||
}
|
||
return blocks, mediaType
|
||
}
|
||
|
||
func (a *Agent) mediaToBlocks(payload map[string]interface{}, mediaType string, source string) ([]agentAPI.ContentBlock, string) {
|
||
data, _ := payload["data"].(string)
|
||
mime, _ := payload["mime"].(string)
|
||
url, _ := payload["url"].(string)
|
||
alt, _ := payload["alt"].(string)
|
||
if alt == "" {
|
||
if source == "" {
|
||
source = "unknown"
|
||
}
|
||
alt = fmt.Sprintf("[从 %s 收到了 %s]", source, mediaType)
|
||
}
|
||
|
||
var blocks []agentAPI.ContentBlock
|
||
|
||
desc := ""
|
||
switch mediaType {
|
||
case "image":
|
||
desc = a.inputCfg.Image.DescribePrompt
|
||
if desc == "" {
|
||
desc = fmt.Sprintf("从 %s 收到了一张图片，请使用 describe_image 工具查看详情。", source)
|
||
}
|
||
case "audio":
|
||
desc = a.inputCfg.Audio.DescribePrompt
|
||
if desc == "" {
|
||
desc = fmt.Sprintf("从 %s 收到了一段音频，请使用 transcribe_audio 工具查看内容。", source)
|
||
}
|
||
}
|
||
blocks = append(blocks, agentAPI.ContentBlock{Type: "text", Text: desc})
|
||
|
||
if data != "" || url != "" {
|
||
imgURL := url
|
||
if data != "" {
|
||
if mime == "" {
|
||
mime = "image/png"
|
||
}
|
||
imgURL = "data:" + mime + ";base64," + data
|
||
}
|
||
if mediaType == "image" {
|
||
blocks = append(blocks, agentAPI.ContentBlock{
|
||
Type: "image_url",
|
||
ImageURL: &agentAPI.ImageURL{URL: imgURL, Detail: "auto"},
|
||
})
|
||
} else if mediaType == "audio" {
|
||
blocks = append(blocks, agentAPI.ContentBlock{
|
||
Type: "audio_url",
|
||
AudioURL: &agentAPI.AudioURL{URL: imgURL},
|
||
})
|
||
}
|
||
}
|
||
|
||
return blocks, alt
|
||
}
|
||
|
||
// emitSkippedReply 给被跳过任务的**同步**调用方一个终态。
|
||
//
|
||
// 为什么要单独一条路径而不是复用 emitResponse：跳过意味着“我们没有处理这条输入”，
|
||
// 不应对外发 agent_output 事件（否则 WebUI 聊天记录会凭空多出一条空消息），
|
||
// 但必须写 ResponseCh——否则 cli/clawhub 这类无超时的同步注入会永久挂起。
|
||
//
|
||
// 非阻塞写：ResponseCh 由同步调用方以 cap=1 创建，调用方超时离开后仍可写入。
|
||
func (a *Agent) emitSkippedReply(evt *agentIO.InputEvent, reason string) {
|
||
if evt == nil || evt.ResponseCh == nil {
|
||
return
|
||
}
|
||
ch := evt.OutputChannel
|
||
if ch == "" {
|
||
ch = evt.Source
|
||
}
|
||
payload := map[string]interface{}{
|
||
"content": "",
|
||
"request_id": evt.RequestID,
|
||
"skipped": true,
|
||
"reason": reason,
|
||
}
|
||
select {
|
||
case evt.ResponseCh <- &agentIO.OutputEvent{
|
||
RequestID: evt.RequestID,
|
||
Target: evt.Source,
|
||
Type: "text",
|
||
Payload: payload,
|
||
Done: true,
|
||
OutputChannel: ch,
|
||
}:
|
||
default:
|
||
}
|
||
}
|
||
|
||
func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
|
||
// 通道一律从**输入事件**推导（内核不持有"当前通道"）。
|
||
ch := outputChannelOf(evt)
|
||
stageCtx := &sdk.StageContext{
|
||
FinalText: response,
|
||
Phase: sdk.StageBeforeOutput,
|
||
Extra: map[string]interface{}{"output_channel": ch},
|
||
}
|
||
a.runStage(sdk.StageBeforeOutput, stageCtx)
|
||
response = stageCtx.FinalText
|
||
|
||
payload := map[string]interface{}{
|
||
"content": response,
|
||
"request_id": evt.RequestID,
|
||
}
|
||
if stageCtx.ReasoningContent != "" {
|
||
payload["reasoning_content"] = stageCtx.ReasoningContent
|
||
}
|
||
if stageCtx.TokenUsage != nil {
|
||
payload["usage"] = stageCtx.TokenUsage
|
||
}
|
||
if evt.ResponseCh != nil {
|
||
// 非阻塞写：ResponseCh 由同步调用方以 cap=1 创建。按不变量 I5（每任务恰一次
|
||
// 终态）这里永远写得进去；但一旦哪天写出第二次，阻塞会卡死**调度器 goroutine**
|
||
// （整个 agent 停摆），而丢弃只是丢一条回执——与 emitSkippedReply 对称。
|
||
select {
|
||
case evt.ResponseCh <- &agentIO.OutputEvent{
|
||
RequestID: evt.RequestID,
|
||
Target: evt.Source,
|
||
Type: "text",
|
||
Payload: payload,
|
||
Done: true,
|
||
OutputChannel: ch,
|
||
}:
|
||
default:
|
||
log.Printf("[agent] ResponseCh 已满，终态回执被丢弃（request=%s，可能违反不变量 I5）", evt.RequestID)
|
||
}
|
||
}
|
||
|
||
out := map[string]interface{}{
|
||
"content": response,
|
||
"channel": ch,
|
||
"source": evt.Source,
|
||
}
|
||
if stageCtx.ReasoningContent != "" {
|
||
out["reasoning_content"] = stageCtx.ReasoningContent
|
||
}
|
||
a.publishEvent(events.EventAgentOutput, out)
|
||
stageCtx.Phase = sdk.StageAfterOutput
|
||
a.runStage(sdk.StageAfterOutput, stageCtx)
|
||
}
|
||
|
||
// pruneOnInput 按声明的上下文策略裁剪上下文，返回归档的事件数。
|
||
//
|
||
// 默认**不裁剪**：ContextPolicy 必须在注入点（payload 的 context_policy）
|
||
// 或通道定义（ChannelDef.ContextPolicy）上显式声明为 prune 才会裁剪。
|
||
//
|
||
// 为什么把无条件裁剪改成需声明：裁剪会把低相关事件归档到文档记忆并从上下文里
|
||
// 移走，是破坏性的。此前每条输入都裁一次，于是「谁把上下文裁了」在排查时无从
|
||
// 得知；而插件注入的内容也会被不相关的内容挤掉。按来源/注入点声明后，触发条件
|
||
// 是可枚举、可审计的。
|
||
//
|
||
// 查询向量取**清洗后**的输入（通道 Cleaner 的输出），与工具侧同一套语义：
|
||
// 原始输入里的 ANSI/base64/JSON 包装会把相关性打分带偏，裁掉本该保留的事件。
|
||
//
|
||
// 实际执行交由 memoryPass（与召回共用入口、query、预算与审计）。
|
||
func (a *Agent) pruneOnInput(evt *agentIO.InputEvent, cleanInput string) int {
|
||
if !a.pruneDeclared(evt) {
|
||
return 0
|
||
}
|
||
return a.memoryPass(cleanInput, "input:"+evt.Source, true, false, sceneKeysFor(evt, "")).Archived
|
||
}
|
||
|
||
// pruneDeclared 判定这次输入是否显式声明了裁剪。
|
||
//
|
||
// 优先级：注入点声明的（payload）> 通道声明的（ChannelDef）> 默认不裁剪。
|
||
// 注入点是更窄的声明面，同一通道下的不同注入可以有不同意图。
|
||
func (a *Agent) pruneDeclared(evt *agentIO.InputEvent) bool {
|
||
if p, ok := evt.Payload["context_policy"].(string); ok && p != "" {
|
||
return p == pubsdk.ContextPolicyPrune
|
||
}
|
||
if a.io != nil {
|
||
if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok {
|
||
return chDef.ContextPolicy == pubsdk.ContextPolicyPrune
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// recallDeclared 判定这次输入是否要触发记忆召回（注入）。
|
||
//
|
||
// 与 pruneDeclared **正交**：prune 管“踢出去”（归档低相关 L0 事件），
|
||
// recall 管“取进来”（把 L2/L3 相关记忆注入本轮）。
|
||
//
|
||
// 默认值与 prune 刻意相反：召回是只读增量、日常对话本就需要，所以**默认 auto**；
|
||
// 只有显式声明 recall_policy=none（如中断通知的 meta 文本）才关闭。
|
||
// 优先级同 prune：注入点（payload）> 通道（ChannelDef）> 默认 auto。
|
||
func (a *Agent) recallDeclared(evt *agentIO.InputEvent) bool {
|
||
if evt == nil {
|
||
return true
|
||
}
|
||
if p, ok := evt.Payload["recall_policy"].(string); ok && p != "" {
|
||
return p != pubsdk.RecallPolicyNone
|
||
}
|
||
if a.io != nil {
|
||
if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.RecallPolicy != "" {
|
||
return chDef.RecallPolicy != pubsdk.RecallPolicyNone
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
// cleanInputFor 解析这条输入在计算层应当使用的清洗文本。
|
||
//
|
||
// 优先级：注入点声明的 cleaner（payload.cleaner_name，引用某个已注册的通道
|
||
// cleaner）> 按 source 查到的通道 cleaner > 原文。
|
||
//
|
||
// 声明的 cleaner 名字查不到时**记日志并回退**，而不是静默当没声明：
|
||
// 注入是 fire-and-forget 的，插件那边看不到错误；至少要在内核日志里留下
|
||
// 「你声明的清洗没生效」的痕迹，否则排查时只能看到「记忆里的内容很脏」。
|
||
func (a *Agent) cleanInputFor(evt *agentIO.InputEvent, input string) string {
|
||
if a.io == nil {
|
||
return input
|
||
}
|
||
if name, ok := evt.Payload["cleaner_name"].(string); ok && name != "" {
|
||
if chDef, ok := a.io.GetInputChannelDef(name); ok && chDef.Cleaner != nil {
|
||
return chDef.Cleaner(input)
|
||
}
|
||
log.Printf("[agent] 注入声明了 cleaner_name=%q 但没有注册过该通道的 Cleaner，已回退", name)
|
||
}
|
||
if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.Cleaner != nil {
|
||
return chDef.Cleaner(input)
|
||
}
|
||
return input
|
||
}
|