mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 01:18:08 +00:00
refactor(scheduler): M1 把 process() 拆成 step 状态机 + TaskFrame（行为等价）
设计依据 docs/zh/input-scheduler-design.md §14 M1。 - 新增 task.go：Step 游标、TaskFrame，以及 stepPrepare/stepLLM/ stepToolBegin/stepToolExec/stepToolAfter/stepTurnEnd 六个 step； 原函数内嵌的 provider 回退/重试抽为 resolveProviders + callLLMWithFallback - process() 改为驱动状态机的薄壳：签名不变，调用方（processInput/ processConsolidation/测试）零改动 - StepToolExec 显式标注为临界区：工具副作用不可回滚，执行中不是安全点 - 步数上限护栏：转移缺失时以错误退出而非死循环 - 新增 task_test.go：R3（工具往返结果与配对正确）、X3（多轮必然终止）、 批中途中断必须放弃剩余工具、未知 step 必须失败退出 - 验收：既有 agent 全量测试通过；-race 通过（TMPDIR 指向真实磁盘， /tmp tmpfs 已 98% 满会导致链接失败，与代码无关）
This commit is contained in:
JianFeeeee
@ -7,12 +7,10 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||||
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
|
||||
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
)
|
||||
|
||||
// continuationPlaceholder 是工具轮之后补的 user 占位内容。
|
||||
@ -99,420 +97,6 @@ func dropContinuationPlaceholders(msgs []agentAPI.Message) []agentAPI.Message {
|
||||
return out
|
||||
}
|
||||
|
||||
func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.provider == nil {
|
||||
return "", nil, nil, fmt.Errorf("agent: no LLM provider configured")
|
||||
}
|
||||
|
||||
budget := ComputeTokenBudget(a.provider, a.systemPrompt)
|
||||
|
||||
memContext := a.buildMemoryContext(input, budget.MemoryTokens)
|
||||
sysPrompt := a.buildSystemPrompt(memContext, input)
|
||||
tools := a.buildToolDefs()
|
||||
|
||||
msgs := a.buildMessages(sysPrompt, input, budget.ContextTokens)
|
||||
// 工具提醒（interrupt）：以 system 角色注入，不让模型误认为用户发言
|
||||
if a.interruptInput {
|
||||
last := msgs[len(msgs)-1]
|
||||
last.Role = "system"
|
||||
last.Content = "[中断消息] " + last.Content
|
||||
msgs[len(msgs)-1] = last
|
||||
a.interruptInput = false
|
||||
}
|
||||
if blocks, ok := stageCtx.Extra["media_blocks"].([]agentAPI.ContentBlock); ok && len(blocks) > 0 {
|
||||
if len(msgs) > 0 {
|
||||
msgs[len(msgs)-1].Blocks = blocks
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[agent] tool call loop start, max_ctx=%d target=%d fixed=%d mem=%d ctx=%d %d tools, %d events, personality=%t, docs=%d",
|
||||
budget.MaxContext, budget.TargetUsage, budget.FixedTokens, budget.MemoryTokens, budget.ContextTokens,
|
||||
len(tools), a.context.Len(),
|
||||
a.personality != nil && a.personality.Content != "",
|
||||
a.docStoreSize())
|
||||
|
||||
if a.runStage(sdk.StagePreAction, stageCtx) {
|
||||
return *stageCtx.Response, toolsUsed, toolResults, nil
|
||||
}
|
||||
if len(stageCtx.ContextMsgs) > 0 {
|
||||
for _, m := range stageCtx.ContextMsgs {
|
||||
role, _ := m["role"].(string)
|
||||
content, _ := m["content"].(string)
|
||||
if role != "" {
|
||||
msgs = append(msgs, agentAPI.Message{Role: role, Content: content})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// lastBatchReplyOnly 记录上一批工具调用是否全部是输出通道发送。
|
||||
lastBatchReplyOnly := false
|
||||
|
||||
for turn := 0; ; turn++ {
|
||||
for _, interrupt := range a.drainInterrupts() {
|
||||
msgs = append(msgs, agentAPI.Message{
|
||||
Role: "system",
|
||||
Content: "[中断消息] " + interrupt,
|
||||
})
|
||||
}
|
||||
|
||||
// zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验),
|
||||
// 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。
|
||||
// 注意:仅当尾部确为工具轮产物(assistant/tool)时才补位;首轮 system 上下文结尾不补,
|
||||
// 否则会错误覆盖实际用户输入(如 injectSourceContext 追加的 system 说明)。
|
||||
//
|
||||
// 补位前先移除前面轮次插入的同类占位，保证占位**不随轮次线性累积**——
|
||||
// 占位是核心插的传输层附加物，不是用户发言，不该在 prompt 里叠成 N 条。
|
||||
//
|
||||
// 文案分情况：上一批全是 output_send__* 时不能说“继续”，详见
|
||||
// replyDeliveredPlaceholder 的说明。
|
||||
msgs = dropContinuationPlaceholders(msgs)
|
||||
if last := msgs[len(msgs)-1]; last.Role == "assistant" || last.Role == "tool" {
|
||||
msgs = append(msgs, agentAPI.Message{
|
||||
Role: "user",
|
||||
Content: continuationFor(lastBatchReplyOnly),
|
||||
})
|
||||
}
|
||||
|
||||
req := &agentAPI.CompletionRequest{
|
||||
Messages: msgs,
|
||||
MaxTokens: 4096,
|
||||
Tools: tools,
|
||||
ToolChoice: "auto",
|
||||
DisableThinking: !a.thinkingEnabled,
|
||||
}
|
||||
|
||||
var providers []agentAPI.Provider
|
||||
if a.providerManager != nil {
|
||||
// 精确模型名走 byModel 路由；AUTO/空走优先级链
|
||||
var allProviders []agentAPI.Provider
|
||||
if req.Model != "" && !strings.EqualFold(req.Model, "AUTO") {
|
||||
allProviders = a.providerManager.ResolveForModel(req.Model)
|
||||
} else {
|
||||
allProviders = a.providerManager.OrderedProviders()
|
||||
}
|
||||
providers = make([]agentAPI.Provider, 0, len(allProviders))
|
||||
for _, p := range allProviders {
|
||||
if a.providerManager.IsAvailable(p.Name()) {
|
||||
providers = append(providers, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(providers) == 0 {
|
||||
providers = []agentAPI.Provider{a.provider}
|
||||
}
|
||||
var resp *agentAPI.CompletionResponse
|
||||
var llmErr error
|
||||
|
||||
for pi, fbProvider := range providers {
|
||||
if pi > 0 {
|
||||
log.Printf("[agent] LLM fallback: trying provider %q (fallback #%d/%d)",
|
||||
fbProvider.Name(), pi, len(providers)-1)
|
||||
}
|
||||
|
||||
// 同源瞬时错误重试：网关瞬断（502/503/504/429/网络抖动）通常秒级恢复，
|
||||
// 直接跳下一个 provider（或直接报错）会丢掉本可成功的请求。
|
||||
// 凭证错误（401/403）与用户中断不重试。
|
||||
const maxAttempts = 2
|
||||
for attempt := 1; attempt <= maxAttempts; attempt++ {
|
||||
if attempt > 1 {
|
||||
log.Printf("[agent] provider %q transient failure, retry %d/%d in 2s: %v",
|
||||
fbProvider.Name(), attempt, maxAttempts, llmErr)
|
||||
select {
|
||||
case <-time.After(2 * time.Second):
|
||||
case <-a.ctx.Done():
|
||||
llmErr = a.ctx.Err()
|
||||
}
|
||||
if llmErr == nil || errors.Is(llmErr, context.Canceled) || errors.Is(llmErr, context.DeadlineExceeded) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
fCtx, fCancel := context.WithCancel(a.ctx)
|
||||
a.llmMu.Lock()
|
||||
a.cancelLLM = fCancel
|
||||
a.llmMu.Unlock()
|
||||
|
||||
resp, llmErr = chatStreamWithFallback(fCtx, fbProvider, req, a)
|
||||
|
||||
a.llmMu.Lock()
|
||||
a.cancelLLM = nil
|
||||
a.llmMu.Unlock()
|
||||
fCancel()
|
||||
|
||||
if llmErr == nil {
|
||||
a.providerManager.ResetAvailability(fbProvider.Name())
|
||||
if fbProvider != a.provider {
|
||||
a.provider = fbProvider
|
||||
log.Printf("[agent] switched active provider to %q after fallback",
|
||||
fbProvider.Name())
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
// 用户中断：立即终止，不重试也不换 provider
|
||||
if errors.Is(llmErr, context.Canceled) {
|
||||
break
|
||||
}
|
||||
// 凭证错误：重试无意义，跳出重试循环进入 provider 标记/切换
|
||||
var pe *agentAPI.ProviderError
|
||||
if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) {
|
||||
break
|
||||
}
|
||||
// 其余错误（含 5xx/429/网络）：还有重试机会则继续，否则跳出
|
||||
}
|
||||
|
||||
if llmErr == nil {
|
||||
break
|
||||
}
|
||||
if errors.Is(llmErr, context.Canceled) {
|
||||
break
|
||||
}
|
||||
var pe *agentAPI.ProviderError
|
||||
if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) {
|
||||
a.providerManager.ReportStatus(fbProvider.Name(), pe.StatusCode)
|
||||
log.Printf("[agent] provider %q marked unavailable (HTTP %d)", fbProvider.Name(), pe.StatusCode)
|
||||
} else {
|
||||
a.providerManager.MarkUnavailable(fbProvider.Name())
|
||||
}
|
||||
log.Printf("[agent] provider %q failed: %v", fbProvider.Name(), llmErr)
|
||||
}
|
||||
|
||||
if llmErr != nil {
|
||||
if errors.Is(llmErr, context.Canceled) && a.ctx.Err() == nil {
|
||||
if a.currentOutputChannel == "_consolidation_" {
|
||||
return "", toolsUsed, toolResults, fmt.Errorf("interrupted by user input")
|
||||
}
|
||||
continue
|
||||
}
|
||||
return "", toolsUsed, toolResults, fmt.Errorf("all %d providers failed, last error: %w",
|
||||
len(providers), llmErr)
|
||||
}
|
||||
|
||||
stageCtx.LLMText = resp.Content
|
||||
stageCtx.ReasoningContent = resp.ReasoningContent
|
||||
stageCtx.TokenUsage = map[string]int{
|
||||
"prompt_tokens": resp.TokenUsage.Prompt,
|
||||
"completion_tokens": resp.TokenUsage.Completion,
|
||||
"total_tokens": resp.TokenUsage.Total,
|
||||
}
|
||||
stageCtx.ToolCalls = convertToolCalls(resp.ToolCalls)
|
||||
for i := range stageCtx.ToolCalls {
|
||||
if stageCtx.ToolCalls[i].Plugin == "" {
|
||||
stageCtx.ToolCalls[i].Plugin = a.resolveToolPlugin(stageCtx.ToolCalls[i].Name)
|
||||
}
|
||||
}
|
||||
if a.runStage(sdk.StagePostAction, stageCtx) {
|
||||
return *stageCtx.Response, toolsUsed, toolResults, nil
|
||||
}
|
||||
resp.Content = stageCtx.LLMText
|
||||
resp.ToolCalls = convertBackToolCalls(stageCtx.ToolCalls)
|
||||
|
||||
chainPayload := map[string]interface{}{
|
||||
"content": resp.Content,
|
||||
"reasoning": resp.ReasoningContent,
|
||||
"tool_calls": resp.ToolCalls,
|
||||
"phase": "intermediate",
|
||||
"turn": turn,
|
||||
}
|
||||
if resp.TokenUsage.Total > 0 {
|
||||
chainPayload["usage"] = map[string]int{
|
||||
"prompt": resp.TokenUsage.Prompt,
|
||||
"completion": resp.TokenUsage.Completion,
|
||||
"total": resp.TokenUsage.Total,
|
||||
}
|
||||
}
|
||||
a.publishEvent(events.EventAgentLLMChain, chainPayload)
|
||||
|
||||
if resp.ReasoningContent != "" {
|
||||
a.publishEvent(events.EventReasoning, map[string]interface{}{
|
||||
"content": resp.ReasoningContent,
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
}
|
||||
|
||||
if len(resp.ToolCalls) == 0 {
|
||||
return resp.Content, toolsUsed, toolResults, nil
|
||||
}
|
||||
|
||||
// 本批是否全部是输出通道发送（=模型刚交付了给用户的回复）。
|
||||
// 必须在执行前判定：执行过程中的中断/拒绝分支会 continue/break，
|
||||
// 放在循环里统计会漏。
|
||||
replyOnly := true
|
||||
for _, tc := range resp.ToolCalls {
|
||||
if !isOutputDeliveryTool(tc.Name) {
|
||||
replyOnly = false
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
contentOnce := true
|
||||
for _, tc := range resp.ToolCalls {
|
||||
if len(a.interceptCh) > 0 {
|
||||
for _, interrupt := range a.drainInterrupts() {
|
||||
msgs = append(msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
|
||||
}
|
||||
a.publishEvent(events.EventToolCall, map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"plugin": a.resolveToolPlugin(tc.Name),
|
||||
"args": tc.Arguments,
|
||||
"status": "interrupted",
|
||||
"reason": "user interrupt before execution",
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
break
|
||||
}
|
||||
|
||||
toolsUsed = append(toolsUsed, tc.Name)
|
||||
pluginName := a.resolveToolPlugin(tc.Name)
|
||||
log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID)
|
||||
if tc.RawArguments != "" {
|
||||
log.Printf("[agent] tool %s raw_arguments: %s", tc.Name, truncateStr(tc.RawArguments, 300))
|
||||
}
|
||||
|
||||
sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Plugin: pluginName, Arguments: tc.Arguments}
|
||||
stageCtx.ToolCalls = []sdk.ToolCall{sdkTC}
|
||||
stageCtx.ToolResults = nil
|
||||
if a.runStage(sdk.StageBeforeToolcall, stageCtx) {
|
||||
result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name)
|
||||
msgs = append(msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}})
|
||||
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
|
||||
a.publishEvent(events.EventToolCall, map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"plugin": pluginName,
|
||||
"args": tc.Arguments,
|
||||
"result": result,
|
||||
"status": "denied",
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
continue
|
||||
}
|
||||
tc.Arguments = stageCtx.ToolCalls[0].Arguments
|
||||
|
||||
if pluginName != "" && !a.pluginHealth.isHealthy(pluginName) {
|
||||
result := fmt.Sprintf("插件 %s 处于崩溃状态，已跳过执行，等待自动恢复重载", pluginName)
|
||||
log.Printf("[agent] skip tool %s: plugin %s unhealthy", tc.Name, pluginName)
|
||||
msgs = append(msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}})
|
||||
msgs = append(msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
|
||||
continue
|
||||
}
|
||||
|
||||
result := a.executeToolCall(tc)
|
||||
toolResults = append(toolResults, ToolResultItem{Name: tc.Name, Output: result})
|
||||
log.Printf("[agent] tool %s result: %s", tc.Name, truncateStr(result, 100))
|
||||
|
||||
stageCtx.ToolResults = []sdk.ToolResult{{CallID: tc.ID, Name: tc.Name, Plugin: pluginName, Success: true, Result: result}}
|
||||
a.runStage(sdk.StageAfterToolcall, stageCtx)
|
||||
if len(stageCtx.ToolResults) > 0 {
|
||||
if r, ok := stageCtx.ToolResults[0].Result.(string); ok {
|
||||
result = r
|
||||
}
|
||||
}
|
||||
// ContextPolicy: prune 工具调用后执行上下文裁剪（§13.8）
|
||||
if def := a.stageHost.ToolDef(tc.Name); def != nil && def.ContextPolicy == "prune" {
|
||||
if a.context != nil {
|
||||
topK := a.maxContextSize - 1
|
||||
if topK < 1 {
|
||||
topK = 1
|
||||
}
|
||||
// 查询向量取**清洗后**的有效内容，否则噪声（ANSI/base64/JSON
|
||||
// 包装）会把相关性打分带偏，裁掉本该保留的事件。
|
||||
a.context.Prune(a.toolOutputForQuery(tc.Name, result), topK, a.docStore)
|
||||
}
|
||||
}
|
||||
|
||||
msgContent := ""
|
||||
if contentOnce {
|
||||
msgContent = resp.Content
|
||||
contentOnce = false
|
||||
}
|
||||
msgs = append(msgs, agentAPI.Message{Role: "assistant", Content: msgContent, ReasoningContent: resp.ReasoningContent, ToolCalls: []agentAPI.ToolCall{tc}})
|
||||
|
||||
// 多模态工具结果：插件通过 SDK.SetToolBlocks 注入 image_url/audio_url block。
|
||||
//
|
||||
// 媒体不挂在 tool message 上，而是另起一条紧随其后的 user message——
|
||||
// 这也是插件文案一直在说的「注入后续对话」。
|
||||
// 为何不能挂 tool message：同一张图、同一模型、三轮实测——
|
||||
// 图在 user message → 3/3 读到
|
||||
// 图在 tool message → 0/3（模型答「没能读到这张图」）
|
||||
// tool 纯文本 + 后接 user → 3/3 读到
|
||||
// tool message 那轮 prompt_tokens 反而更高（7967 vs 7089），base64 确实
|
||||
// 进了上游，但 role=tool 上的多模态 content 数组不被当作可视内容。
|
||||
//
|
||||
// 主模型不支持该模态时更不能直接塞：网关会把 image_url 静默剥离后仍
|
||||
// 返回 200，模型回答「我没有看到图片」而内核以为注入成功。改走回退链。
|
||||
toolMsg := agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}
|
||||
var mediaMsg *agentAPI.Message
|
||||
if rawBlocks := a.io.ConsumeToolBlocks(); len(rawBlocks) > 0 {
|
||||
var blocks []agentAPI.ContentBlock
|
||||
for _, b := range rawBlocks {
|
||||
if cb, ok := b.(pubsdk.ContentBlock); ok {
|
||||
// 跨包类型拷贝（pubsdk.ContentBlock → agentAPI.ContentBlock）
|
||||
block := agentAPI.ContentBlock{Type: cb.Type, Text: cb.Text}
|
||||
if cb.ImageURL != nil {
|
||||
block.ImageURL = &agentAPI.ImageURL{URL: cb.ImageURL.URL, Detail: cb.ImageURL.Detail}
|
||||
}
|
||||
if cb.AudioURL != nil {
|
||||
block.AudioURL = &agentAPI.AudioURL{URL: cb.AudioURL.URL}
|
||||
}
|
||||
blocks = append(blocks, block)
|
||||
}
|
||||
}
|
||||
if len(blocks) > 0 {
|
||||
// 先落进 CAS：无论下面走直视还是回退转写，媒体本体都该进记忆。
|
||||
// 不存的后果是 ToolResultItem.Output 只剩那句
|
||||
// "[已将图片注入后续对话] /tmp/x.png"，文件一删线索就断了。
|
||||
a.stageMediaDigests(a.captureBlockMedia(blocks, tc.Name)...)
|
||||
|
||||
if native, fallbackText := a.prepareToolBlocks(blocks); len(native) > 0 {
|
||||
// 能直视：另起一条 user message 承载媒体，并补一句来源说明，
|
||||
// 否则模型会把它当成用户新发的图而不是工具拉回来的。
|
||||
mediaBlocks := append([]agentAPI.ContentBlock{{
|
||||
Type: "text",
|
||||
Text: fmt.Sprintf("[以下是 %s 注入的媒体内容]", tc.Name),
|
||||
}}, native...)
|
||||
mediaMsg = &agentAPI.Message{Role: "user", Blocks: mediaBlocks}
|
||||
} else if fallbackText != "" {
|
||||
// 回退链已把媒体转写成文字：并进 tool message 的纯文本 content，
|
||||
// 不再另起消息（文字在 tool message 里本来就能被读到）。
|
||||
toolMsg.Content = result + "\n\n" + fallbackText
|
||||
result = toolMsg.Content
|
||||
if len(toolResults) > 0 {
|
||||
toolResults[len(toolResults)-1].Output = result
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
msgs = append(msgs, toolMsg)
|
||||
if mediaMsg != nil {
|
||||
// 必须紧跟在 toolMsg 之后：中间插入其他消息会让 tool_call_id 配对断开。
|
||||
msgs = append(msgs, *mediaMsg)
|
||||
}
|
||||
|
||||
a.publishEvent(events.EventToolCall, map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"plugin": pluginName,
|
||||
"args": tc.Arguments,
|
||||
"result": result,
|
||||
"status": "ok",
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
|
||||
if len(a.interceptCh) > 0 {
|
||||
for _, interrupt := range a.drainInterrupts() {
|
||||
msgs = append(msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// 供下一轮顶部选择补位文案。
|
||||
lastBatchReplyOnly = replyOnly
|
||||
}
|
||||
}
|
||||
|
||||
// chatStreamWithFallback 优先流式调用 provider，失败时回退非流式 Chat()。
|
||||
//
|
||||
// 流式路径：ChatStream 拿到 chunk channel，逐块累积 content/reasoning_content，
|
||||
|
||||
613
internal/agent/core/task.go
Normal file
613
internal/agent/core/task.go
Normal file
@ -0,0 +1,613 @@
|
||||
package core
|
||||
|
||||
// 任务状态机（M1：行为等价的纯重构）。
|
||||
//
|
||||
// 背景与设计见 docs/zh/input-scheduler-design.md。
|
||||
//
|
||||
// M1 只做一件事：把原先「一个 425 行的 process() 大函数」拆成
|
||||
// **显式 step 游标 + TaskFrame**。目的不是加能力，而是让「现场」变成数据——
|
||||
// 之后 M3 才能把帧存进 suspendPool 并在安全点恢复。
|
||||
//
|
||||
// 行为等价的判据：既有全部 agent 测试通过，且 R3/X3（见设计文档 §11）通过。
|
||||
//
|
||||
// 本文件**不引入**优先级、抢占、队列与并发；那些在 M2 起逐层加上。
|
||||
//
|
||||
// Step 与安全点（设计文档 §4.2）：
|
||||
// step 与 step 之间是安全点；StepToolExec（工具执行）与 StepPrepare 中的
|
||||
// ONNX/落盘片段是**临界区**，执行中不可抢占。
|
||||
//
|
||||
// 与设计文档的差异：文档里的 StepBeforeOutput / StepAfterOutput / StepCommit /
|
||||
// StepFinish 属于 emitResponse 与 processInput 层，M1 不动它们（M6 再迁入帧）。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||||
"gitcode.com/JianFeeeee/HomeAgent/internal/events"
|
||||
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
)
|
||||
|
||||
// Step 是任务状态机的游标。
|
||||
type Step int
|
||||
|
||||
const (
|
||||
// StepPrepare 构建消息与工具表、应用中断标记、跑 pre_action 阶段。
|
||||
StepPrepare Step = iota
|
||||
// StepLLM 轮次顶部（中断/占位）+ LLM 调用（含 provider 回退与重试）+ post_action。
|
||||
StepLLM
|
||||
// StepToolBegin 取本批下一个工具，跑 before_toolcall；被拒/插件不健康则跳过。
|
||||
StepToolBegin
|
||||
// StepToolExec 执行工具。**临界区**：副作用不可回滚，执行中不是安全点。
|
||||
StepToolExec
|
||||
// StepToolAfter after_toolcall 阶段、上下文裁剪、消息与事件组装、批后中断检查。
|
||||
StepToolAfter
|
||||
// StepTurnEnd 收尾本批并进入下一轮。
|
||||
StepTurnEnd
|
||||
)
|
||||
|
||||
// stepOutcome 是一次 step 执行的结果。
|
||||
type stepOutcome int
|
||||
|
||||
const (
|
||||
// outcomeContinue 继续执行下一个 step（游标可能停在原地以表达"重跑本 step"）。
|
||||
outcomeContinue stepOutcome = iota
|
||||
// outcomeDone 任务成功结束，响应在 frame.Response。
|
||||
outcomeDone
|
||||
// outcomeFailed 任务失败结束，错误在 frame.Err。
|
||||
outcomeFailed
|
||||
)
|
||||
|
||||
// TaskFrame 承载一个任务在安全点之间必须存活的所有状态。
|
||||
//
|
||||
// 不变量（设计文档 §8.1 I3）：帧是**纯数据**；不得持有任何锁或资源跨越安全点。
|
||||
type TaskFrame struct {
|
||||
Input string
|
||||
StageCtx *sdk.StageContext
|
||||
|
||||
// 跨轮次状态
|
||||
Msgs []agentAPI.Message
|
||||
Tools []interface{}
|
||||
ToolsUsed []string
|
||||
ToolResults []ToolResultItem
|
||||
Turn int
|
||||
LastBatchReplyOnly bool
|
||||
|
||||
// 当前工具批
|
||||
PendingTools []agentAPI.ToolCall
|
||||
ToolIdx int
|
||||
ReplyOnly bool
|
||||
ContentOnce bool
|
||||
CurTool agentAPI.ToolCall
|
||||
CurToolPlugin string
|
||||
CurResult string
|
||||
Resp *agentAPI.CompletionResponse
|
||||
|
||||
// 游标与终态
|
||||
Step Step
|
||||
Response string
|
||||
Err error
|
||||
}
|
||||
|
||||
func (a *Agent) newTaskFrame(input string, stageCtx *sdk.StageContext) *TaskFrame {
|
||||
return &TaskFrame{Input: input, StageCtx: stageCtx, Step: StepPrepare}
|
||||
}
|
||||
|
||||
// process 驱动状态机直到任务结束，返回与原实现完全相同的四元组。
|
||||
//
|
||||
// 保留该签名是为了让 M1 成为纯内部重构：所有调用方（processInput /
|
||||
// processConsolidation / 测试）无需改动。
|
||||
func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
|
||||
if a.provider == nil {
|
||||
return "", nil, nil, fmt.Errorf("agent: no LLM provider configured")
|
||||
}
|
||||
|
||||
f := a.newTaskFrame(input, stageCtx)
|
||||
// 步数上限只是防"转移缺失导致死循环"的护栏；正常任务远达不到。
|
||||
const maxSteps = 1 << 20
|
||||
for i := 0; i < maxSteps; i++ {
|
||||
switch a.step(f) {
|
||||
case outcomeDone:
|
||||
return f.Response, f.ToolsUsed, f.ToolResults, nil
|
||||
case outcomeFailed:
|
||||
return "", f.ToolsUsed, f.ToolResults, f.Err
|
||||
}
|
||||
}
|
||||
return "", f.ToolsUsed, f.ToolResults,
|
||||
fmt.Errorf("agent: task step budget exhausted（状态机未收敛，疑似转移缺失）")
|
||||
}
|
||||
|
||||
// step 执行恰好一个 step。
|
||||
func (a *Agent) step(f *TaskFrame) stepOutcome {
|
||||
switch f.Step {
|
||||
case StepPrepare:
|
||||
return a.stepPrepare(f)
|
||||
case StepLLM:
|
||||
return a.stepLLM(f)
|
||||
case StepToolBegin:
|
||||
return a.stepToolBegin(f)
|
||||
case StepToolExec:
|
||||
return a.stepToolExec(f)
|
||||
case StepToolAfter:
|
||||
return a.stepToolAfter(f)
|
||||
case StepTurnEnd:
|
||||
return a.stepTurnEnd(f)
|
||||
default:
|
||||
f.Err = fmt.Errorf("agent: unknown task step %d", f.Step)
|
||||
return outcomeFailed
|
||||
}
|
||||
}
|
||||
|
||||
// stepPrepare 构建本轮任务的初始帧。
|
||||
func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome {
|
||||
budget := ComputeTokenBudget(a.provider, a.systemPrompt)
|
||||
|
||||
memContext := a.buildMemoryContext(f.Input, budget.MemoryTokens)
|
||||
sysPrompt := a.buildSystemPrompt(memContext, f.Input)
|
||||
f.Tools = a.buildToolDefs()
|
||||
|
||||
f.Msgs = a.buildMessages(sysPrompt, f.Input, budget.ContextTokens)
|
||||
// 工具提醒（interrupt）：以 system 角色注入，不让模型误认为用户发言
|
||||
if a.interruptInput {
|
||||
last := f.Msgs[len(f.Msgs)-1]
|
||||
last.Role = "system"
|
||||
last.Content = "[中断消息] " + last.Content
|
||||
f.Msgs[len(f.Msgs)-1] = last
|
||||
a.interruptInput = false
|
||||
}
|
||||
if blocks, ok := f.StageCtx.Extra["media_blocks"].([]agentAPI.ContentBlock); ok && len(blocks) > 0 {
|
||||
if len(f.Msgs) > 0 {
|
||||
f.Msgs[len(f.Msgs)-1].Blocks = blocks
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[agent] tool call loop start, max_ctx=%d target=%d fixed=%d mem=%d ctx=%d %d tools, %d events, personality=%t, docs=%d",
|
||||
budget.MaxContext, budget.TargetUsage, budget.FixedTokens, budget.MemoryTokens, budget.ContextTokens,
|
||||
len(f.Tools), a.context.Len(),
|
||||
a.personality != nil && a.personality.Content != "",
|
||||
a.docStoreSize())
|
||||
|
||||
if a.runStage(sdk.StagePreAction, f.StageCtx) {
|
||||
f.Response = *f.StageCtx.Response
|
||||
return outcomeDone
|
||||
}
|
||||
if len(f.StageCtx.ContextMsgs) > 0 {
|
||||
for _, m := range f.StageCtx.ContextMsgs {
|
||||
role, _ := m["role"].(string)
|
||||
content, _ := m["content"].(string)
|
||||
if role != "" {
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: role, Content: content})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
f.Step = StepLLM
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
// stepLLM 是轮次顶部与 LLM 调用。
|
||||
//
|
||||
// 取消（context.Canceled 且 agent 未退出）时**留在本 step 并 Turn++**——等价于
|
||||
// 原实现的 `continue`：重新排空中断、补占位、重新请求。抢占挂起将在 M3 从这里接管。
|
||||
func (a *Agent) stepLLM(f *TaskFrame) stepOutcome {
|
||||
for _, interrupt := range a.drainInterrupts() {
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{
|
||||
Role: "system",
|
||||
Content: "[中断消息] " + interrupt,
|
||||
})
|
||||
}
|
||||
|
||||
// zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验),
|
||||
// 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。
|
||||
f.Msgs = dropContinuationPlaceholders(f.Msgs)
|
||||
if last := f.Msgs[len(f.Msgs)-1]; last.Role == "assistant" || last.Role == "tool" {
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{
|
||||
Role: "user",
|
||||
Content: continuationFor(f.LastBatchReplyOnly),
|
||||
})
|
||||
}
|
||||
|
||||
req := &agentAPI.CompletionRequest{
|
||||
Messages: f.Msgs,
|
||||
MaxTokens: 4096,
|
||||
Tools: f.Tools,
|
||||
ToolChoice: "auto",
|
||||
DisableThinking: !a.thinkingEnabled,
|
||||
}
|
||||
|
||||
providers := a.resolveProviders(req)
|
||||
resp, llmErr := a.callLLMWithFallback(req, providers)
|
||||
|
||||
if llmErr != nil {
|
||||
if errors.Is(llmErr, context.Canceled) && a.ctx.Err() == nil {
|
||||
if a.currentOutputChannel == "_consolidation_" {
|
||||
f.Err = fmt.Errorf("interrupted by user input")
|
||||
return outcomeFailed
|
||||
}
|
||||
f.Turn++
|
||||
return outcomeContinue // 重跑 StepLLM
|
||||
}
|
||||
f.Err = fmt.Errorf("all %d providers failed, last error: %w", len(providers), llmErr)
|
||||
return outcomeFailed
|
||||
}
|
||||
|
||||
f.StageCtx.LLMText = resp.Content
|
||||
f.StageCtx.ReasoningContent = resp.ReasoningContent
|
||||
f.StageCtx.TokenUsage = map[string]int{
|
||||
"prompt_tokens": resp.TokenUsage.Prompt,
|
||||
"completion_tokens": resp.TokenUsage.Completion,
|
||||
"total_tokens": resp.TokenUsage.Total,
|
||||
}
|
||||
f.StageCtx.ToolCalls = convertToolCalls(resp.ToolCalls)
|
||||
for i := range f.StageCtx.ToolCalls {
|
||||
if f.StageCtx.ToolCalls[i].Plugin == "" {
|
||||
f.StageCtx.ToolCalls[i].Plugin = a.resolveToolPlugin(f.StageCtx.ToolCalls[i].Name)
|
||||
}
|
||||
}
|
||||
if a.runStage(sdk.StagePostAction, f.StageCtx) {
|
||||
f.Response = *f.StageCtx.Response
|
||||
return outcomeDone
|
||||
}
|
||||
resp.Content = f.StageCtx.LLMText
|
||||
resp.ToolCalls = convertBackToolCalls(f.StageCtx.ToolCalls)
|
||||
|
||||
chainPayload := map[string]interface{}{
|
||||
"content": resp.Content,
|
||||
"reasoning": resp.ReasoningContent,
|
||||
"tool_calls": resp.ToolCalls,
|
||||
"phase": "intermediate",
|
||||
"turn": f.Turn,
|
||||
}
|
||||
if resp.TokenUsage.Total > 0 {
|
||||
chainPayload["usage"] = map[string]int{
|
||||
"prompt": resp.TokenUsage.Prompt,
|
||||
"completion": resp.TokenUsage.Completion,
|
||||
"total": resp.TokenUsage.Total,
|
||||
}
|
||||
}
|
||||
a.publishEvent(events.EventAgentLLMChain, chainPayload)
|
||||
|
||||
if resp.ReasoningContent != "" {
|
||||
a.publishEvent(events.EventReasoning, map[string]interface{}{
|
||||
"content": resp.ReasoningContent,
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
}
|
||||
|
||||
if len(resp.ToolCalls) == 0 {
|
||||
f.Response = resp.Content
|
||||
return outcomeDone
|
||||
}
|
||||
|
||||
// 本批是否全部是输出通道发送（=模型刚交付了给用户的回复）。
|
||||
// 必须在执行前判定：执行过程中的中断/拒绝分支会 continue/break，放在循环里统计会漏。
|
||||
f.Resp = resp
|
||||
f.ReplyOnly = true
|
||||
for _, tc := range resp.ToolCalls {
|
||||
if !isOutputDeliveryTool(tc.Name) {
|
||||
f.ReplyOnly = false
|
||||
break
|
||||
}
|
||||
}
|
||||
f.ContentOnce = true
|
||||
f.PendingTools = resp.ToolCalls
|
||||
f.ToolIdx = 0
|
||||
f.Step = StepToolBegin
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
// stepToolBegin 取本批下一个工具；批已耗尽或发生中断则进入收尾。
|
||||
func (a *Agent) stepToolBegin(f *TaskFrame) stepOutcome {
|
||||
if f.ToolIdx >= len(f.PendingTools) {
|
||||
f.Step = StepTurnEnd
|
||||
return outcomeContinue
|
||||
}
|
||||
tc := f.PendingTools[f.ToolIdx]
|
||||
|
||||
if len(a.interceptCh) > 0 {
|
||||
for _, interrupt := range a.drainInterrupts() {
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
|
||||
}
|
||||
a.publishEvent(events.EventToolCall, map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"plugin": a.resolveToolPlugin(tc.Name),
|
||||
"args": tc.Arguments,
|
||||
"status": "interrupted",
|
||||
"reason": "user interrupt before execution",
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
f.Step = StepTurnEnd
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
f.ToolsUsed = append(f.ToolsUsed, tc.Name)
|
||||
pluginName := a.resolveToolPlugin(tc.Name)
|
||||
log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID)
|
||||
if tc.RawArguments != "" {
|
||||
log.Printf("[agent] tool %s raw_arguments: %s", tc.Name, truncateStr(tc.RawArguments, 300))
|
||||
}
|
||||
|
||||
sdkTC := sdk.ToolCall{ID: tc.ID, Name: tc.Name, Plugin: pluginName, Arguments: tc.Arguments}
|
||||
f.StageCtx.ToolCalls = []sdk.ToolCall{sdkTC}
|
||||
f.StageCtx.ToolResults = nil
|
||||
if a.runStage(sdk.StageBeforeToolcall, f.StageCtx) {
|
||||
result := fmt.Sprintf("工具 %s 已被插件拒绝", tc.Name)
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}})
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
|
||||
a.publishEvent(events.EventToolCall, map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"plugin": pluginName,
|
||||
"args": tc.Arguments,
|
||||
"result": result,
|
||||
"status": "denied",
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
f.ToolIdx++
|
||||
return outcomeContinue
|
||||
}
|
||||
tc.Arguments = f.StageCtx.ToolCalls[0].Arguments
|
||||
|
||||
if pluginName != "" && !a.pluginHealth.isHealthy(pluginName) {
|
||||
result := fmt.Sprintf("插件 %s 处于崩溃状态，已跳过执行，等待自动恢复重载", pluginName)
|
||||
log.Printf("[agent] skip tool %s: plugin %s unhealthy", tc.Name, pluginName)
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "assistant", ToolCalls: []agentAPI.ToolCall{tc}})
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result})
|
||||
f.ToolIdx++
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
f.CurTool = tc
|
||||
f.CurToolPlugin = pluginName
|
||||
f.Step = StepToolExec
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
// stepToolExec 执行工具。**临界区**：见设计文档 §4.3。
|
||||
func (a *Agent) stepToolExec(f *TaskFrame) stepOutcome {
|
||||
result := a.executeToolCall(f.CurTool)
|
||||
f.CurResult = result
|
||||
f.ToolResults = append(f.ToolResults, ToolResultItem{Name: f.CurTool.Name, Output: result})
|
||||
log.Printf("[agent] tool %s result: %s", f.CurTool.Name, truncateStr(result, 100))
|
||||
|
||||
f.StageCtx.ToolResults = []sdk.ToolResult{{
|
||||
CallID: f.CurTool.ID, Name: f.CurTool.Name, Plugin: f.CurToolPlugin,
|
||||
Success: true, Result: result,
|
||||
}}
|
||||
f.Step = StepToolAfter
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
// stepToolAfter 是工具执行后的全部后处理（阶段、裁剪、消息与事件）。
|
||||
func (a *Agent) stepToolAfter(f *TaskFrame) stepOutcome {
|
||||
tc := f.CurTool
|
||||
pluginName := f.CurToolPlugin
|
||||
result := f.CurResult
|
||||
|
||||
a.runStage(sdk.StageAfterToolcall, f.StageCtx)
|
||||
if len(f.StageCtx.ToolResults) > 0 {
|
||||
if r, ok := f.StageCtx.ToolResults[0].Result.(string); ok {
|
||||
result = r
|
||||
}
|
||||
}
|
||||
// ContextPolicy: prune 工具调用后执行上下文裁剪（§13.8）
|
||||
if def := a.stageHost.ToolDef(tc.Name); def != nil && def.ContextPolicy == "prune" {
|
||||
if a.context != nil {
|
||||
topK := a.maxContextSize - 1
|
||||
if topK < 1 {
|
||||
topK = 1
|
||||
}
|
||||
// 查询向量取**清洗后**的有效内容，否则噪声（ANSI/base64/JSON 包装）
|
||||
// 会把相关性打分带偏，裁掉本该保留的事件。
|
||||
a.context.Prune(a.toolOutputForQuery(tc.Name, result), topK, a.docStore)
|
||||
}
|
||||
}
|
||||
|
||||
msgContent := ""
|
||||
if f.ContentOnce {
|
||||
msgContent = f.Resp.Content
|
||||
f.ContentOnce = false
|
||||
}
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{
|
||||
Role: "assistant", Content: msgContent,
|
||||
ReasoningContent: f.Resp.ReasoningContent,
|
||||
ToolCalls: []agentAPI.ToolCall{tc},
|
||||
})
|
||||
|
||||
// 多模态工具结果：插件通过 SDK.SetToolBlocks 注入 image_url/audio_url block。
|
||||
//
|
||||
// 媒体不挂在 tool message 上，而是另起一条紧随其后的 user message——
|
||||
// 这也是插件文案一直在说的「注入后续对话」。
|
||||
// 为何不能挂 tool message：同一张图、同一模型、三轮实测——
|
||||
// 图在 user message → 3/3 读到
|
||||
// 图在 tool message → 0/3（模型答「没能读到这张图」）
|
||||
// tool 纯文本 + 后接 user → 3/3 读到
|
||||
// tool message 那轮 prompt_tokens 反而更高（7967 vs 7089），base64 确实
|
||||
// 进了上游，但 role=tool 上的多模态 content 数组不被当作可视内容。
|
||||
//
|
||||
// 主模型不支持该模态时更不能直接塞：网关会把 image_url 静默剥离后仍
|
||||
// 返回 200，模型回答「我没有看到图片」而内核以为注入成功。改走回退链。
|
||||
toolMsg := agentAPI.Message{Role: "tool", ToolCallID: tc.ID, Content: result}
|
||||
var mediaMsg *agentAPI.Message
|
||||
if rawBlocks := a.io.ConsumeToolBlocks(); len(rawBlocks) > 0 {
|
||||
var blocks []agentAPI.ContentBlock
|
||||
for _, b := range rawBlocks {
|
||||
if cb, ok := b.(pubsdk.ContentBlock); ok {
|
||||
// 跨包类型拷贝（pubsdk.ContentBlock → agentAPI.ContentBlock）
|
||||
block := agentAPI.ContentBlock{Type: cb.Type, Text: cb.Text}
|
||||
if cb.ImageURL != nil {
|
||||
block.ImageURL = &agentAPI.ImageURL{URL: cb.ImageURL.URL, Detail: cb.ImageURL.Detail}
|
||||
}
|
||||
if cb.AudioURL != nil {
|
||||
block.AudioURL = &agentAPI.AudioURL{URL: cb.AudioURL.URL}
|
||||
}
|
||||
blocks = append(blocks, block)
|
||||
}
|
||||
}
|
||||
if len(blocks) > 0 {
|
||||
// 先落进 CAS：无论下面走直视还是回退转写，媒体本体都该进记忆。
|
||||
a.stageMediaDigests(a.captureBlockMedia(blocks, tc.Name)...)
|
||||
|
||||
if native, fallbackText := a.prepareToolBlocks(blocks); len(native) > 0 {
|
||||
// 能直视：另起一条 user message 承载媒体，并补一句来源说明。
|
||||
mediaBlocks := append([]agentAPI.ContentBlock{{
|
||||
Type: "text",
|
||||
Text: fmt.Sprintf("[以下是 %s 注入的媒体内容]", tc.Name),
|
||||
}}, native...)
|
||||
mediaMsg = &agentAPI.Message{Role: "user", Blocks: mediaBlocks}
|
||||
} else if fallbackText != "" {
|
||||
toolMsg.Content = result + "\n\n" + fallbackText
|
||||
result = toolMsg.Content
|
||||
if len(f.ToolResults) > 0 {
|
||||
f.ToolResults[len(f.ToolResults)-1].Output = result
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
f.Msgs = append(f.Msgs, toolMsg)
|
||||
if mediaMsg != nil {
|
||||
// 必须紧跟在 toolMsg 之后：中间插入其他消息会让 tool_call_id 配对断开。
|
||||
f.Msgs = append(f.Msgs, *mediaMsg)
|
||||
}
|
||||
|
||||
a.publishEvent(events.EventToolCall, map[string]interface{}{
|
||||
"tool": tc.Name,
|
||||
"plugin": pluginName,
|
||||
"args": tc.Arguments,
|
||||
"result": result,
|
||||
"status": "ok",
|
||||
"channel": a.currentOutputChannel,
|
||||
})
|
||||
|
||||
f.ToolIdx++
|
||||
if len(a.interceptCh) > 0 {
|
||||
for _, interrupt := range a.drainInterrupts() {
|
||||
f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
|
||||
}
|
||||
f.Step = StepTurnEnd
|
||||
return outcomeContinue
|
||||
}
|
||||
f.Step = StepToolBegin
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
// stepTurnEnd 收尾本批并进入下一轮。
|
||||
func (a *Agent) stepTurnEnd(f *TaskFrame) stepOutcome {
|
||||
// 供下一轮顶部选择补位文案。
|
||||
f.LastBatchReplyOnly = f.ReplyOnly
|
||||
f.Turn++
|
||||
f.Step = StepLLM
|
||||
return outcomeContinue
|
||||
}
|
||||
|
||||
// resolveProviders 按请求模型解析候选 provider（保持原语义）。
|
||||
func (a *Agent) resolveProviders(req *agentAPI.CompletionRequest) []agentAPI.Provider {
|
||||
var providers []agentAPI.Provider
|
||||
if a.providerManager != nil {
|
||||
var allProviders []agentAPI.Provider
|
||||
if req.Model != "" && !strings.EqualFold(req.Model, "AUTO") {
|
||||
allProviders = a.providerManager.ResolveForModel(req.Model)
|
||||
} else {
|
||||
allProviders = a.providerManager.OrderedProviders()
|
||||
}
|
||||
providers = make([]agentAPI.Provider, 0, len(allProviders))
|
||||
for _, p := range allProviders {
|
||||
if a.providerManager.IsAvailable(p.Name()) {
|
||||
providers = append(providers, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(providers) == 0 {
|
||||
providers = []agentAPI.Provider{a.provider}
|
||||
}
|
||||
return providers
|
||||
}
|
||||
|
||||
// callLLMWithFallback 在候选 provider 间回退，并把同源瞬时错误重试一次。
|
||||
// 逐行等价于原 process() 内的双层循环。
|
||||
func (a *Agent) callLLMWithFallback(req *agentAPI.CompletionRequest, providers []agentAPI.Provider) (*agentAPI.CompletionResponse, error) {
|
||||
var resp *agentAPI.CompletionResponse
|
||||
var llmErr error
|
||||
|
||||
for pi, fbProvider := range providers {
|
||||
if pi > 0 {
|
||||
log.Printf("[agent] LLM fallback: trying provider %q (fallback #%d/%d)",
|
||||
fbProvider.Name(), pi, len(providers)-1)
|
||||
}
|
||||
|
||||
// 同源瞬时错误重试：网关瞬断（502/503/504/429/网络抖动）通常秒级恢复，
|
||||
// 直接跳下一个 provider（或直接报错）会丢掉本可成功的请求。
|
||||
// 凭证错误（401/403）与用户中断不重试。
|
||||
const maxAttempts = 2
|
||||
for attempt := 1; attempt <= maxAttempts; attempt++ {
|
||||
if attempt > 1 {
|
||||
log.Printf("[agent] provider %q transient failure, retry %d/%d in 2s: %v",
|
||||
fbProvider.Name(), attempt, maxAttempts, llmErr)
|
||||
select {
|
||||
case <-time.After(2 * time.Second):
|
||||
case <-a.ctx.Done():
|
||||
llmErr = a.ctx.Err()
|
||||
}
|
||||
if llmErr == nil || errors.Is(llmErr, context.Canceled) || errors.Is(llmErr, context.DeadlineExceeded) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
fCtx, fCancel := context.WithCancel(a.ctx)
|
||||
a.llmMu.Lock()
|
||||
a.cancelLLM = fCancel
|
||||
a.llmMu.Unlock()
|
||||
|
||||
resp, llmErr = chatStreamWithFallback(fCtx, fbProvider, req, a)
|
||||
|
||||
a.llmMu.Lock()
|
||||
a.cancelLLM = nil
|
||||
a.llmMu.Unlock()
|
||||
fCancel()
|
||||
|
||||
if llmErr == nil {
|
||||
a.providerManager.ResetAvailability(fbProvider.Name())
|
||||
if fbProvider != a.provider {
|
||||
a.provider = fbProvider
|
||||
log.Printf("[agent] switched active provider to %q after fallback", fbProvider.Name())
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
// 用户中断：立即终止，不重试也不换 provider
|
||||
if errors.Is(llmErr, context.Canceled) {
|
||||
break
|
||||
}
|
||||
// 凭证错误：重试无意义，跳出重试循环进入 provider 标记/切换
|
||||
var pe *agentAPI.ProviderError
|
||||
if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) {
|
||||
break
|
||||
}
|
||||
// 其余错误（含 5xx/429/网络）：还有重试机会则继续，否则跳出
|
||||
}
|
||||
|
||||
if llmErr == nil {
|
||||
break
|
||||
}
|
||||
if errors.Is(llmErr, context.Canceled) {
|
||||
break
|
||||
}
|
||||
var pe *agentAPI.ProviderError
|
||||
if errors.As(llmErr, &pe) && (pe.StatusCode == 401 || pe.StatusCode == 403) {
|
||||
a.providerManager.ReportStatus(fbProvider.Name(), pe.StatusCode)
|
||||
log.Printf("[agent] provider %q marked unavailable (HTTP %d)", fbProvider.Name(), pe.StatusCode)
|
||||
} else {
|
||||
a.providerManager.MarkUnavailable(fbProvider.Name())
|
||||
}
|
||||
log.Printf("[agent] provider %q failed: %v", fbProvider.Name(), llmErr)
|
||||
}
|
||||
|
||||
return resp, llmErr
|
||||
}
|
||||
209
internal/agent/core/task_test.go
Normal file
209
internal/agent/core/task_test.go
Normal file
@ -0,0 +1,209 @@
|
||||
package core
|
||||
|
||||
// M1 验收测试：状态机与 TaskFrame 的**行为等价性**。
|
||||
//
|
||||
// 设计依据 docs/zh/input-scheduler-design.md §11（R3 / X3 的 M1 形态）：
|
||||
// M1 不引入抢占，因此 R3 退化为「经状态机跑出的结果与脚本预期一致」；
|
||||
// X3 在 M1 退化为「驱动循环必然以一次终态返回结束（不空转、不超步数）」。
|
||||
//
|
||||
// 抢占/挂起/恢复/优先级在 M3 起才有测试。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
|
||||
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
|
||||
sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
|
||||
)
|
||||
|
||||
// scriptProvider 按脚本依次返回 CompletionResponse。
|
||||
//
|
||||
// ChatStream 故意返回错误：驱动 chatStreamWithFallback 走非流式回退，
|
||||
// 这样脚本就是「第 N 次调用返回第 N 个响应」，不依赖流式分片语义。
|
||||
type scriptProvider struct {
|
||||
script []*agentAPI.CompletionResponse
|
||||
idx int
|
||||
reqs []*agentAPI.CompletionRequest
|
||||
}
|
||||
|
||||
func (s *scriptProvider) Name() string { return "script" }
|
||||
func (s *scriptProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) {
|
||||
s.reqs = append(s.reqs, req)
|
||||
if s.idx >= len(s.script) {
|
||||
return &agentAPI.CompletionResponse{Content: ""}, nil
|
||||
}
|
||||
r := s.script[s.idx]
|
||||
s.idx++
|
||||
return r, nil
|
||||
}
|
||||
func (s *scriptProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) {
|
||||
return nil, errors.New("script provider: streaming disabled")
|
||||
}
|
||||
func (s *scriptProvider) MaxContextTokens() int { return 8192 }
|
||||
|
||||
func newTaskTestAgent(t *testing.T, sp agentAPI.Provider, sh *StageHost) *Agent {
|
||||
t.Helper()
|
||||
return New(AgentConfig{
|
||||
ID: "task-test",
|
||||
Provider: sp,
|
||||
ProviderManager: agentAPI.NewProviderManager(),
|
||||
StageHost: sh,
|
||||
IO: agentIO.NewIOManager(),
|
||||
})
|
||||
}
|
||||
|
||||
func tc(id, name string) agentAPI.ToolCall {
|
||||
return agentAPI.ToolCall{ID: id, Name: name, Arguments: map[string]interface{}{"q": id}}
|
||||
}
|
||||
|
||||
// R3（M1 形态）：一次工具轮 + 一次收尾轮，结果与工具调用计数必须正确。
|
||||
func TestTaskFrame_R3_ToolRoundTrip(t *testing.T) {
|
||||
sh := NewStageHost()
|
||||
var got []string
|
||||
sh.RegisterTool("t_echo", sdk.ToolDef{Name: "t_echo", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
|
||||
got = append(got, args["q"].(string))
|
||||
return "OUT", nil
|
||||
})
|
||||
|
||||
sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
|
||||
{Content: "让我调用工具", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_echo")}},
|
||||
{Content: "最终答复"},
|
||||
}}
|
||||
a := newTaskTestAgent(t, sp, sh)
|
||||
|
||||
stageCtx := a.stageCtxFromInput("你好", "", "")
|
||||
resp, toolsUsed, toolResults, err := a.process("你好", stageCtx)
|
||||
if err != nil {
|
||||
t.Fatalf("process 返回错误: %v", err)
|
||||
}
|
||||
if resp != "最终答复" {
|
||||
t.Fatalf("响应=%q，期望 %q", resp, "最终答复")
|
||||
}
|
||||
if len(toolsUsed) != 1 || toolsUsed[0] != "t_echo" {
|
||||
t.Fatalf("toolsUsed=%v，期望恰好一次 t_echo", toolsUsed)
|
||||
}
|
||||
if len(toolResults) != 1 || toolResults[0].Name != "t_echo" || toolResults[0].Output != "OUT" {
|
||||
t.Fatalf("toolResults=%+v，期望一条 t_echo/OUT", toolResults)
|
||||
}
|
||||
if len(got) != 1 || got[0] != "c1" {
|
||||
t.Fatalf("工具实参=%v，期望恰好执行一次且参数来自脚本", got)
|
||||
}
|
||||
if len(sp.reqs) != 2 {
|
||||
t.Fatalf("LLM 调用次数=%d，期望 2（工具轮 + 收尾轮）", len(sp.reqs))
|
||||
}
|
||||
|
||||
// 第二轮请求必须携带 assistant(tool_call) + tool 结果两条消息。
|
||||
msgs := sp.reqs[1].Messages
|
||||
var hasAssistantCall, hasToolResult bool
|
||||
for _, m := range msgs {
|
||||
if m.Role == "assistant" && len(m.ToolCalls) == 1 && m.ToolCalls[0].ID == "c1" {
|
||||
hasAssistantCall = true
|
||||
}
|
||||
if m.Role == "tool" && m.ToolCallID == "c1" && m.Content == "OUT" {
|
||||
hasToolResult = true
|
||||
}
|
||||
}
|
||||
if !hasAssistantCall || !hasToolResult {
|
||||
t.Fatalf("第二轮请求缺少工具调用配对：assistant=%v tool=%v", hasAssistantCall, hasToolResult)
|
||||
}
|
||||
}
|
||||
|
||||
// X3（M1 形态）：多轮脚本必须在有限步内以一次终态返回结束。
|
||||
func TestTaskFrame_X3_TerminatesWithinBudget(t *testing.T) {
|
||||
sh := NewStageHost()
|
||||
sh.RegisterTool("t_noop", sdk.ToolDef{Name: "t_noop", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
|
||||
return "ok", nil
|
||||
})
|
||||
|
||||
// 3 个工具轮 + 收尾轮：状态机会在 StepLLM/StepToolBegin/.../StepTurnEnd 间往返 4 次。
|
||||
var script []*agentAPI.CompletionResponse
|
||||
for i := 0; i < 3; i++ {
|
||||
script = append(script, &agentAPI.CompletionResponse{
|
||||
Content: "round",
|
||||
ToolCalls: []agentAPI.ToolCall{tc("c"+string(rune('a'+i)), "t_noop")},
|
||||
})
|
||||
}
|
||||
script = append(script, &agentAPI.CompletionResponse{Content: "done"})
|
||||
|
||||
sp := &scriptProvider{script: script}
|
||||
a := newTaskTestAgent(t, sp, sh)
|
||||
|
||||
resp, toolsUsed, toolResults, err := a.process("跑三轮", a.stageCtxFromInput("跑三轮", "", ""))
|
||||
if err != nil {
|
||||
t.Fatalf("process 返回错误: %v", err)
|
||||
}
|
||||
if resp != "done" {
|
||||
t.Fatalf("响应=%q，期望 done", resp)
|
||||
}
|
||||
if len(toolsUsed) != 3 || len(toolResults) != 3 {
|
||||
t.Fatalf("toolsUsed=%d toolResults=%d，期望各 3", len(toolsUsed), len(toolResults))
|
||||
}
|
||||
// 步数护栏未触发（触发了会是 "step budget exhausted" 错误）。
|
||||
if len(sp.reqs) != 4 {
|
||||
t.Fatalf("LLM 调用次数=%d，期望 4", len(sp.reqs))
|
||||
}
|
||||
}
|
||||
|
||||
// 中断（interceptCh）在工具批中途到达时：本批**剩余工具被放弃**，直接进入下一轮。
|
||||
//
|
||||
// 这是原实现的 `break` 语义（process.go 旧版工具循环尾部），必须保持。
|
||||
func TestTaskFrame_InterruptAbandonsRemainingBatch(t *testing.T) {
|
||||
sh := NewStageHost()
|
||||
var a *Agent
|
||||
sh.RegisterTool("t_first", sdk.ToolDef{Name: "t_first", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
|
||||
// 工具执行期间产生一次中断（模拟插件在工具里注入打断）。
|
||||
a.interceptCh <- &agentIO.InputEvent{Source: "t", Payload: map[string]interface{}{"content": "新的用户输入"}}
|
||||
return "first-out", nil
|
||||
})
|
||||
sh.RegisterTool("t_second", sdk.ToolDef{Name: "t_second", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
|
||||
t.Fatal("批内第二个工具不应被执行：中断必须放弃剩余批次")
|
||||
return nil, nil
|
||||
})
|
||||
|
||||
sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
|
||||
{Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_first"), tc("c2", "t_second")}},
|
||||
{Content: "处理完中断后的答复"},
|
||||
}}
|
||||
a = newTaskTestAgent(t, sp, sh)
|
||||
|
||||
resp, toolsUsed, toolResults, err := a.process("开始", a.stageCtxFromInput("开始", "", ""))
|
||||
if err != nil {
|
||||
t.Fatalf("process 返回错误: %v", err)
|
||||
}
|
||||
if resp != "处理完中断后的答复" {
|
||||
t.Fatalf("响应=%q", resp)
|
||||
}
|
||||
if len(toolsUsed) != 1 || toolsUsed[0] != "t_first" {
|
||||
t.Fatalf("toolsUsed=%v，期望只有 t_first", toolsUsed)
|
||||
}
|
||||
if len(toolResults) != 1 || toolResults[0].Output != "first-out" {
|
||||
t.Fatalf("toolResults=%+v，期望只有 t_first 的结果", toolResults)
|
||||
}
|
||||
|
||||
// 第二轮请求里必须出现被打断内容（system 角色、[中断消息] 前缀）。
|
||||
found := false
|
||||
for _, m := range sp.reqs[1].Messages {
|
||||
if m.Role == "system" && len(m.Content) > 0 && m.Content[0:len("[中断消息]")] == "[中断消息]" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("第二轮请求缺少 [中断消息] system 消息")
|
||||
}
|
||||
}
|
||||
|
||||
// 状态机对未知 step 必须失败退出而不是空转。
|
||||
func TestTaskFrame_UnknownStepFails(t *testing.T) {
|
||||
sp := &scriptProvider{}
|
||||
a := newTaskTestAgent(t, sp, NewStageHost())
|
||||
f := a.newTaskFrame("x", a.stageCtxFromInput("x", "", ""))
|
||||
f.Step = Step(999)
|
||||
if out := a.step(f); out != outcomeFailed {
|
||||
t.Fatalf("未知 step 应返回 outcomeFailed，实际 %v", out)
|
||||
}
|
||||
if f.Err == nil {
|
||||
t.Fatal("未知 step 必须带错误信息")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user