From 38a034bc607cc83c28953490ad71bbd5633f653b Mon Sep 17 00:00:00 2001 From: liut Date: Mon, 13 Jul 2026 15:38:27 +0800 Subject: [PATCH] refactor(agent): extract AgentLoop, add parallel tool execution and terminate --- .gitignore | 1 + ...-13-001-feat-agent-loop-extraction-plan.md | 273 ++++++++ pkg/services/agent/agent_loop.go | 133 ++++ pkg/services/agent/agent_loop_test.go | 627 ++++++++++++++++++ pkg/services/agent/tool_executor.go | 229 +++++++ pkg/services/agent/tool_executor_test.go | 300 +++++++++ pkg/services/llm/event.go | 7 +- pkg/services/tools/registry.go | 11 + pkg/web/api/agent.go | 71 +- pkg/web/api/api.go | 5 +- pkg/web/api/convo_basic.go | 17 - pkg/web/api/handle_convo.go | 328 +++------ pkg/web/api/handle_convo_test.go | 35 - pkg/web/api/handle_platform.go | 160 ++--- pkg/web/api/tool_executor.go | 112 ---- 15 files changed, 1760 insertions(+), 549 deletions(-) create mode 100644 docs/plans/2026-07-13-001-feat-agent-loop-extraction-plan.md create mode 100644 pkg/services/agent/agent_loop.go create mode 100644 pkg/services/agent/agent_loop_test.go create mode 100644 pkg/services/agent/tool_executor.go create mode 100644 pkg/services/agent/tool_executor_test.go delete mode 100644 pkg/web/api/tool_executor.go diff --git a/.gitignore b/.gitignore index e2958d4..d9bfe9a 100644 --- a/.gitignore +++ b/.gitignore @@ -112,6 +112,7 @@ tests/ .idea/ +.deepseek/ *.local.* morrigan diff --git a/docs/plans/2026-07-13-001-feat-agent-loop-extraction-plan.md b/docs/plans/2026-07-13-001-feat-agent-loop-extraction-plan.md new file mode 100644 index 0000000..ff2d991 --- /dev/null +++ b/docs/plans/2026-07-13-001-feat-agent-loop-extraction-plan.md @@ -0,0 +1,273 @@ +--- +title: Agent 循环解耦与工具执行增强 +type: refactor +status: active +date: 2026-07-13 +origin: docs/brainstorms/2026-07-13-agent-loop-extraction-requirements.md +--- + +# Agent 循环解耦与工具执行增强 + +## Summary + +抽取三处 Handler 中重复的 Agent 循环为独立的 `AgentLoop`，通过 `iter.Seq2` 输出事件解耦传输层；同步加入工具并行执行（goroutine fan-out）、terminate 提前退出、beforeToolCall/afterToolCall 钩子。新建 `pkg/services/agent` 包，扩展 `ToolExecutor`，重构三个 Handler 为纯事件消费者。 + +--- + +## Problem Frame + +`handle_convo.go`（SSE）、`handle_platform.go`（平台频道 Channel，如 WeCom/Feishu）、`agent.go`（CLI）各有一套相同的循环模板：调 LLM → 收集 tool calls → 串行执行工具 → 追加结果 → loop。修改循环行为需要改三处，工具执行无并行能力。 + +需求文档 `docs/brainstorms/2026-07-13-agent-loop-extraction-requirements.md` 已完成范围确认，本计划是其实现规划。 + +--- + +## Requirements + +- R1. 创建 `pkg/services/agent` 包，`AgentLoop` 封装 LLM 调用 + 工具执行 + 循环迭代（origin R1） +- R2. 通过 `iter.Seq2[*llm.Event, error]` 输出事件，无传输层依赖（origin R2） +- R3. 支持流式（`StreamChat`）和非流式（`Chat`）两种模式（origin R3） +- R4. `maxLoopIterations` 保留为安全网（origin R4） +- R5. 三处 Handler 替换为 AgentLoop 消费（origin R5） +- R6. 多个 tool calls 并发执行（origin R6） +- R7. 工具结果保持 LLM 返回顺序（origin R7） +- R8. `ToolResult.Terminate` 支持整批提前退出（origin R8） +- R9. `BeforeToolCall` 钩子（origin R9） +- R10. `AfterToolCall` 钩子（origin R10） + +> R11 `ShouldStopAfterTurn` 钩子已延期至后续实现，详见 [Deferred to Follow-Up Work](#deferred-to-follow-up-work)。 + +**Origin actors:** A1 (HTTP SSE Handler), A2 (Channel Handler), A3 (CLI Agent), A4 (LLM Provider), A5 (Tool Registry) +**Origin flows:** F1 (Agent 循环流式), F2 (工具并行执行), F3 (terminate 提前退出), F4 (工具钩子拦截) +**Origin acceptance examples:** AE1 (SSE 行为不变), AE2 (两工具并发), AE3 (terminate 退出), AE4 (before hook block), AE5 (after hook 审计) + +--- + +## Scope Boundaries + +- AgentMessage 分层抽象 +- Steering / Follow-up 运行时消息注入 +- 事件体系细化为多类型 +- Pluggable transport + +### Deferred to Follow-Up Work + +- R11 `shouldStopAfterTurn`: 规划阶段确认 Channel handler 场景后单独实现 + +--- + +## Context & Research + +### Relevant Code and Patterns + +- `pkg/web/api/handle_convo.go`: SSE 流式循环（`chatStreamResponseLoop` + `doChatStream`，~100 行循环逻辑） +- `pkg/web/api/handle_platform.go`: Channel 流式循环（`handleStreamingReply` + `doChannelStream`，~70 行循环逻辑） +- `pkg/web/api/agent.go`: CLI 流式循环（`Run` 方法，~50 行循环逻辑） +- `pkg/web/api/tool_executor.go`: `ToolExecutor` 结构体及 `ExecuteToolCallLoop` / `ExecuteToolCalls` 方法 +- `pkg/services/llm/event.go`: `Event` 结构体及 `ToolResult` 子结构 +- `pkg/services/llm/client.go`: `Client` 接口（`Chat` / `StreamChat`） +- `pkg/services/llm/types.go`: `Message`、`ToolCall`、`ToolDefinition` 类型 + +### External References + +- pi-mono `@mariozechner/pi-agent-core` 的 Agent 类、AgentLoop 引擎、工具并行执行和 terminate 机制作为设计参考 + +--- + +## Key Technical Decisions + +- **AgentLoop 输出 `iter.Seq2`**: `Run()` 方法返回 `iter.Seq2[*llm.Event, error]`，与 LLM 的 `StreamChat` 保持一致的迭代器语义。调用方通过 `for event, err := range loop.Run(ctx, messages, tools)` 消费，无需 goroutine 桥接或 channel 缓冲策略 +- **ToolExecutor 迁入 agent 包**: `ToolExecutor` 从 `pkg/web/api` 迁至 `pkg/services/agent`，消除 services→web 跨层反向依赖。原有位置保留 type alias 或直接删除，调用方更新 import 路径 +- **ToolExecutor 零侵入扩展**: 新增 `BeforeToolCall` 和 `AfterToolCall` 为可选 func 字段，`ExecuteToolCalls` 内部在关键点调用。不设置这些字段时行为与当前完全一致 +- **并行执行编排**: 在 `ExecuteToolCalls` 内部，用 `[]promise` 模式——预分配结果切片、每个工具调用的 goroutine 写入固定索引位置、`sync.WaitGroup` 等待全部完成。保证 LLM 返回顺序的同时并发执行 +- **Terminate 放在 `ToolResult` 上**: 新增 `Terminate bool` 字段到 `llm.ToolResult`。工具通过 `Registry.Invoke` 返回的 `map[string]any` 中可选的 `"terminate"` 键（`bool` 类型）设值。提取协议：若 map 含 `"terminate": true` 则 `ToolResult.Terminate = true`，否则为 `false`。`AgentLoop` 在工具批量执行后检查所有结果是否都设了 terminate +- **非流式循环保留兼容路径**: `ExecuteToolCallLoop` 本次重构中保留以供向后兼容，后续可择机废弃。新增 `AgentLoop.RunNonStreaming` 方法作为非流式场景的首选路径 + +--- + +## Open Questions + +### Resolved During Planning + +- AgentLoop 输出 `iter.Seq2`，调用方通过 `for event, err := range` 消费，无需 goroutine 桥接 +- 工具并行上限：默认不设上限，`ExecuteToolCalls` 内部全部并发。后续如需要可加 `maxConcurrency` 参数 +- `Terminate` 字段放在 `llm.ToolResult` 上，工具实现方通过返回值设标记 +- `handle_platform.go` 流控适配（平台频道 Channel）：`StartStream` → 首次收到 AgentLoop 事件时调用；`AppendStream` → 每次收到 delta 事件时调用；`FinishStream` → AgentLoop 迭代结束时调用。流控状态维护在 Handler 侧，AgentLoop 不感知 + +### Deferred to Implementation + +- `ShouldStopAfterTurn` 钩子的接口签名和生效时机 +- 并行执行的 goroutine panic recovery 策略 +- `ExecuteToolCallLoop` 的废弃时间表 + +--- + +## Output Structure + +``` +pkg/services/agent/ # 新建 +├── agent_loop.go # AgentLoop 结构体 + Run() / RunNonStreaming() +├── tool_executor.go # 从 pkg/web/api 迁入: 加钩子字段、并行执行、terminate 检查 +├── agent_loop_test.go # 单元测试 + +pkg/services/llm/event.go # 修改: ToolResult 加 Terminate 字段 +pkg/web/api/tool_executor.go # 删除: ToolExecutor 已迁至 pkg/services/agent +pkg/web/api/handle_convo.go # 修改: 循环替换为 AgentLoop +pkg/web/api/handle_platform.go # 修改: 循环替换为 AgentLoop +pkg/web/api/agent.go # 修改: Run 方法替换为 AgentLoop +``` + +--- + +## Implementation Units + +### U1. 扩展 ToolExecutor：钩子 + 并行 + terminate + +**Goal:** 扩展 `ExecuteToolCalls`：加入 before/after 钩子调用、goroutine 并行执行、及 terminate 标记检测。返回值增加 `allTerminate bool`，现有调用方同步适配。 + +**Requirements:** R6, R7, R8, R9, R10 + +**Dependencies:** None + +**Files:** +- Create: `pkg/services/agent/tool_executor.go`（从 `pkg/web/api/tool_executor.go` 迁入并扩展） +- Modify: `pkg/services/llm/event.go`（`ToolResult` 加 `Terminate`） +- Modify: `pkg/web/api/tool_executor.go`（替换为 re-export 或删除，调用方改为 import 新位置） + +**Approach:** +- `ToolExecutor` 新增两个可选字段：`BeforeToolCall`（`func(ctx, name string, params map[string]any) (block bool, reason string)`）和 `AfterToolCall`（`func(ctx, name string, result map[string]any) map[string]any`） +- `ExecuteToolCalls` 内部重构：预分配切片、每个 tool call 启动 `go func()`（含 `defer wg.Done()` + `defer recover()` 防 panic 死锁）、WaitGroup 等待、按原始索引收集结果 +- 结果收集后检查所有 `ToolResult.Terminate`，若全部为 true 则返回 `allTerminate=true` +- 当 `BeforeToolCall` block 工具时，生成的 error `ToolResult` 默认 `Terminate=false`，不阻塞正常 terminate 退出 +- `ExecuteToolCalls` 返回值增加 `allTerminate bool`，现有调用方同步适配（共 3 处：`tool_executor.go` 的 `ExecuteToolCallLoop` 内部调用需捕获 `allTerminate`、`handle_convo.go` 的 `executeToolCallLoop`、`agent.go` 的 `Run`） +- 搬迁时需同步处理跨包依赖：`formatToolResult` 从 `handle_convo.go` 迁入本文件，`logger()` 替换为 `slog.Default()` + +**Execution note:** 先写单元测试覆盖并发 + 钩子 + terminate 场景，再改实现 + +**Test scenarios:** +- Happy path: 两个工具调用，goroutine 并发执行，结果按调用顺序返回 +- Happy path: BeforeToolCall 返回 block=false，工具正常执行 +- Happy path: AfterToolCall 修改结果并返回 +- Edge case: BeforeToolCall 返回 block=true，工具不执行，生成错误 tool result +- Edge case: 所有工具返回 terminate=true，allTerminate 标记为 true +- Edge case: 部分工具 terminate，allTerminate 为 false +- Edge case: 无 tool calls 时直接返回空结果 + +--- + +### U2. 创建 AgentLoop + +**Goal:** 在 `pkg/services/agent` 中创建 `AgentLoop`，封装流式和非流式的完整 Agent 循环逻辑。 + +**Requirements:** R1, R2, R3, R4 + +**Dependencies:** U1 + +**Files:** +- Create: `pkg/services/agent/agent_loop.go` +- Create: `pkg/services/agent/agent_loop_test.go` + +**Approach:** +- `AgentLoopConfig` 结构体：`LLM llm.Client`、`ToolExec *ToolExecutor`、`MaxLoop int`、`Stream bool` +- `AgentLoop.Run(ctx, messages, tools) iter.Seq2[*llm.Event, error]`：返回迭代器，调用方通过 `for event, err := range` 消费。内部复用 `StreamChat` 的 `iter.Seq2` 语义，不引入 goroutine 桥接 +- 流式模式：调 `StreamChat`，将每个 chunk 包装为 Event 发送；done 时收集 tool calls，调用 `ExecuteToolCalls`，将 tool result event 发送，若 allTerminate 则退出，否则下一轮 +- 非流式模式：`RunNonStreaming(ctx, messages, tools) (string, error)`——调 `Chat`，收集 tool calls，执行工具，循环直到无 tool calls +- `maxLoopIterations` 检查保留在循环开头，超限时发一条 error event 然后退出 + +**Patterns to follow:** +- `pkg/services/runner/runner.go` 的简洁结构体 + option 模式 +- `pkg/web/api/agent.go` 的 `Run` 方法循环逻辑 + +**Test scenarios:** +- Happy path: 无工具调用的单轮对话，事件序列正确 +- Happy path: 有工具调用的多轮循环，事件包含 tool result +- Happy path: 非流式模式，返回最终答案 +- Edge case: 达到 maxLoop 上限，优雅退出 +- Edge case: 工具执行后 allTerminate=true，循环退出不发起新 LLM 调用 +- Edge case: context cancel 时迭代正常终止 +- Edge case: StreamChat 返回错误时迭代器输出 error 事件然后终止 + +--- + +### U3. 迁移 SSE Handler + +**Goal:** `handle_convo.go` 的 `chatStreamResponseLoop` + `doChatStream` 替换为 AgentLoop 消费。 + +**Requirements:** R5 + +**Dependencies:** U2 + +**Files:** +- Modify: `pkg/web/api/handle_convo.go` + +**Approach:** +- 创建 `AgentLoop`，`for event, err := range loop.Run(ctx, messages, tools)` 消费事件 +- delta 事件 → SSE writeEvent +- tool call 事件 → SSE writeEvent（tool_calls 信息） +- tool result 事件 → SSE writeEvent（可选，当前未向客户端发送 tool result） +- 循环结束 → 发送 done 事件 + 持久化（`Runner.Persist`） +- 删除 `chatStreamResponseLoop`、`doChatStream`、`chatResponse` 结构体及相关的 `doChatStream` 内部字段 +- `chatRequest` 结构体精简，移除不再需要的字段（`hi`、`chunkIdx`、`prompt` 中可合并的部分） + +**Patterns to follow:** +- 保持现有 SSE 事件格式不变（`eventsource.WriteEvent`） +- `writeEvent` 辅助函数保留 +- `Runner.Persist` 调用位置从 `chatStreamResponseLoop` 末尾移到 `for range` 循环外 + +**Test scenarios:** +- 回归：SSE 聊天请求产生的事件序列与重构前一致 +- 回归：usage 记录和 history 持久化正常 + +--- + +### U4. 迁移 Channel Handler + +**Goal:** `handle_platform.go` 的 `handleStreamingReply` + `doChannelStream` 替换为 AgentLoop 消费（Channel 指 WeCom/Feishu 等平台频道，非 Go 语言 channel）。 + +**Requirements:** R5 + +**Dependencies:** U2 + +**Files:** +- Modify: `pkg/web/api/handle_platform.go` + +**Approach:** +- 创建 `AgentLoop`，`for event, err := range loop.Run(ctx, messages, tools)` 消费事件 +- 首次 delta → 调 `sr.StartStream` +- 后续 delta → 调 `sr.AppendStream`（累加模式，WeCom 覆盖语义） +- 迭代结束 → 调 `sr.FinishStream` + 持久化 +- 删除 `handleStreamingReply`、`doChannelStream` 中的循环逻辑 +- `buildChatMessagesAndTools` 保留 + +**Patterns to follow:** +- WeCom 的 AppendStream 累加模式：Handler 侧维护 `contentBuilder`，每次 delta 追加后全量发送 +- 错误处理：StreamChat 错误 → `sr.FinishStream` 发送错误消息 + +**Test scenarios:** +- 回归：Channel 聊天流式行为与重构前一致 +- Edge case: LLM 返回错误时 FinishStream 正确发送错误消息 + +--- + +### U5. 迁移 CLI Agent + +**Goal:** `agent.go` 的 `Run` 方法替换为 AgentLoop 调用。 + +**Requirements:** R5 + +**Dependencies:** U2 + +**Files:** +- Modify: `pkg/web/api/agent.go` + +**Approach:** +- `Run` 方法内创建 `AgentLoop`，`for event, err := range agentLoop.Run(ctx, messages, tools)` 消费事件 +- delta 事件 → yield(event, nil) +- tool result 事件 → yield(event, nil) +- 迭代结束 → return +- `StreamChat` 方法无需修改——它已包装 `Run`，`Run` 替换为 `AgentLoop.Run` 后自动生效 +- `Chat` 方法改为调用 `AgentLoop.RunNonStreaming` + +**Test scenarios:** +- 回归：CLI 流式输出与重构前一致 +- 回归：非流式 Chat 返回正确的完整答案 diff --git a/pkg/services/agent/agent_loop.go b/pkg/services/agent/agent_loop.go new file mode 100644 index 0000000..22a3fe1 --- /dev/null +++ b/pkg/services/agent/agent_loop.go @@ -0,0 +1,133 @@ +package agent + +import ( + "context" + "fmt" + "iter" + "strings" + + "github.com/liut/morign/pkg/services/llm" +) + +// AgentLoopOption is a functional option for AgentLoopConfig. +type AgentLoopOption func(*AgentLoopConfig) + +// WithMaxLoop sets the maximum number of loop iterations before safety-net termination. +func WithMaxLoop(n int) AgentLoopOption { + return func(c *AgentLoopConfig) { + c.MaxLoop = n + } +} + +// AgentLoopConfig holds configuration for AgentLoop. +type AgentLoopConfig struct { + LLM llm.Client // LLM client for Chat and StreamChat + ToolExec *ToolExecutor // tool executor (same package) + MaxLoop int // safety net: max loop iterations, default 5 +} + +// AgentLoop encapsulates the full agent loop: LLM call + tool execution + iteration. +type AgentLoop struct { + cfg AgentLoopConfig +} + +// NewAgentLoop creates a new AgentLoop with the given config and options. +// Defaults MaxLoop to 5 when unset or <= 0. +func NewAgentLoop(cfg AgentLoopConfig, opts ...AgentLoopOption) *AgentLoop { + if cfg.MaxLoop <= 0 { + cfg.MaxLoop = 5 + } + for _, opt := range opts { + opt(&cfg) + } + return &AgentLoop{cfg: cfg} +} + +// Run executes the streaming agent loop, returning an iter.Seq2 that yields +// events from StreamChat and tool execution. The caller consumes events via +// `for event, err := range loop.Run(ctx, messages, tools)`. +// +// The iterator runs synchronously within the caller's goroutine. It stops when: +// - LLM returns no tool calls (final answer) +// - all executed tools signal terminate=true +// - maxLoop iterations are exceeded (yields error event) +// - StreamChat returns an error (yields error event) +// - the caller breaks out of the range loop +func (al *AgentLoop) Run(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) iter.Seq2[*llm.Event, error] { + return func(yield func(*llm.Event, error) bool) { + for iter := 0; iter < al.cfg.MaxLoop; iter++ { + var roundThink strings.Builder + var toolCalls []llm.ToolCall + + // Consume stream events from LLM + for event, err := range al.cfg.LLM.StreamChat(ctx, messages, tools) { + if err != nil { + yield(nil, fmt.Errorf("stream chat: %w", err)) + return + } + roundThink.WriteString(event.Think) + + if !yield(event, nil) { + return // caller stopped consuming + } + + if event.Done { + toolCalls = event.ToolCalls + } + } + + // No tool calls → final answer delivered, we're done + if len(toolCalls) == 0 { + return + } + + // Execute tool calls and yield tool result events + events, updatedMsgs, allTerminate := al.cfg.ToolExec.ExecuteToolCalls( + ctx, messages, toolCalls, roundThink.String(), + ) + messages = updatedMsgs + for _, ev := range events { + if !yield(ev, nil) { + return // caller stopped consuming + } + } + + // All tools requested termination → no more LLM calls needed + if allTerminate { + return + } + } + + // Safety net: max loop iterations exceeded + yield(nil, fmt.Errorf("max loop iterations (%d) exceeded", al.cfg.MaxLoop)) + } +} + +// RunNonStreaming executes the non-streaming agent loop, returning the final +// answer string. It calls Chat, collects tool calls, executes tools, and loops +// until no tool calls remain or all tools signal terminate. +func (al *AgentLoop) RunNonStreaming(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, error) { + for iter := 0; iter < al.cfg.MaxLoop; iter++ { + result, err := al.cfg.LLM.Chat(ctx, messages, tools) + if err != nil { + return "", fmt.Errorf("chat: %w", err) + } + + // No tool calls → return the content as final answer + if len(result.ToolCalls) == 0 { + return result.Content, nil + } + + // Execute tool calls + _, updatedMsgs, allTerminate := al.cfg.ToolExec.ExecuteToolCalls( + ctx, messages, result.ToolCalls, result.Thinking, + ) + messages = updatedMsgs + + if allTerminate { + return result.Content, nil + } + } + + return "", fmt.Errorf("max loop iterations (%d) exceeded", al.cfg.MaxLoop) +} diff --git a/pkg/services/agent/agent_loop_test.go b/pkg/services/agent/agent_loop_test.go new file mode 100644 index 0000000..560958d --- /dev/null +++ b/pkg/services/agent/agent_loop_test.go @@ -0,0 +1,627 @@ +package agent + +import ( + "context" + "errors" + "iter" + "sync/atomic" + "testing" + "time" + + "github.com/liut/morign/pkg/services/llm" +) + +// --------------------------------------------------------------------------- +// mock LLM client helpers +// --------------------------------------------------------------------------- + +// singleSeqMock returns the same event sequence on every StreamChat call. +type singleSeqMock struct { + seq iter.Seq2[*llm.Event, error] +} + +func (m *singleSeqMock) Chat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (*llm.ChatResult, error) { + return nil, errors.New("chat not implemented in mock") +} +func (m *singleSeqMock) StreamChat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) iter.Seq2[*llm.Event, error] { + return m.seq +} +func (m *singleSeqMock) Generate(ctx context.Context, prompt string) (string, *llm.Usage, error) { + return "", nil, errors.New("generate not implemented") +} +func (m *singleSeqMock) Embedding(ctx context.Context, texts []string) ([]float64, error) { + return nil, errors.New("embedding not implemented") +} + +// multiSeqMock returns a different sequence on each StreamChat call (round-robin). +type multiSeqMock struct { + rounds []iter.Seq2[*llm.Event, error] + idx int +} + +func (m *multiSeqMock) Chat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (*llm.ChatResult, error) { + return nil, errors.New("chat not implemented in mock") +} +func (m *multiSeqMock) StreamChat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) iter.Seq2[*llm.Event, error] { + if m.idx >= len(m.rounds) { + return func(yield func(*llm.Event, error) bool) {} + } + seq := m.rounds[m.idx] + m.idx++ + return seq +} +func (m *multiSeqMock) Generate(ctx context.Context, prompt string) (string, *llm.Usage, error) { + return "", nil, errors.New("generate not implemented") +} +func (m *multiSeqMock) Embedding(ctx context.Context, texts []string) ([]float64, error) { + return nil, errors.New("embedding not implemented") +} + +// chatMock returns a fixed ChatResult on every Chat call. +type chatMock struct { + result *llm.ChatResult + err error +} + +func (m *chatMock) Chat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (*llm.ChatResult, error) { + return m.result, m.err +} +func (m *chatMock) StreamChat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) iter.Seq2[*llm.Event, error] { + return func(yield func(*llm.Event, error) bool) {} +} +func (m *chatMock) Generate(ctx context.Context, prompt string) (string, *llm.Usage, error) { + return "", nil, errors.New("generate not implemented") +} +func (m *chatMock) Embedding(ctx context.Context, texts []string) ([]float64, error) { + return nil, errors.New("embedding not implemented") +} + +// multiChatMock returns a different ChatResult on each Chat call. +type multiChatMock struct { + results []*llm.ChatResult + idx int +} + +func (m *multiChatMock) Chat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (*llm.ChatResult, error) { + if m.idx >= len(m.results) { + return &llm.ChatResult{}, nil + } + r := m.results[m.idx] + m.idx++ + return r, nil +} +func (m *multiChatMock) StreamChat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) iter.Seq2[*llm.Event, error] { + return func(yield func(*llm.Event, error) bool) {} +} +func (m *multiChatMock) Generate(ctx context.Context, prompt string) (string, *llm.Usage, error) { + return "", nil, errors.New("generate not implemented") +} +func (m *multiChatMock) Embedding(ctx context.Context, texts []string) ([]float64, error) { + return nil, errors.New("embedding not implemented") +} + +// --------------------------------------------------------------------------- +// event constructors +// --------------------------------------------------------------------------- + +func deltaEv(delta string) *llm.Event { + return &llm.Event{ID: llm.NewEventID(), Timestamp: time.Now(), Author: "assistant", Delta: delta} +} + +func doneEv(toolCalls []llm.ToolCall) *llm.Event { + return &llm.Event{ + ID: llm.NewEventID(), + Timestamp: time.Now(), + Author: "assistant", + Done: true, + ToolCalls: toolCalls, + } +} + +// collectEvents consumes an iter.Seq2 and returns all events + last error. +func collectEvents(seq iter.Seq2[*llm.Event, error]) (events []*llm.Event, lastErr error) { + for ev, err := range seq { + if err != nil { + return events, err + } + events = append(events, ev) + } + return events, nil +} + +// --------------------------------------------------------------------------- +// helper: simple tool that echoes +// --------------------------------------------------------------------------- + +func echoResult(text string) map[string]any { + return map[string]any{"content": []any{map[string]any{"text": text}}} +} + +func echoTerminateResult(text string) map[string]any { + return map[string]any{"content": []any{map[string]any{"text": text}}, "terminate": true} +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 1 — single turn, no tool calls +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_SingleTurnNoTools(t *testing.T) { + mock := &singleSeqMock{seq: func(yield func(*llm.Event, error) bool) { + yield(deltaEv("Hello"), nil) + yield(deltaEv(" world"), nil) + yield(doneEv(nil), nil) // Done with no tool calls + }} + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(testRegistry()), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + events, err := collectEvents(loop.Run(context.Background(), nil, nil)) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) != 3 { + t.Fatalf("expected 3 events, got %d", len(events)) + } + if events[0].Delta != "Hello" { + t.Errorf("event[0].Delta = %q, want Hello", events[0].Delta) + } + if events[1].Delta != " world" { + t.Errorf("event[1].Delta = %q, want ' world'", events[1].Delta) + } + if !events[2].Done { + t.Error("event[2].Done should be true") + } +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 2 — multi-turn with tool calls +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_MultiTurnWithTools(t *testing.T) { + reg := testRegistry() + reg.RegisterInvoker("weather", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return echoResult("sunny, 22°C"), nil + }) + + // Round 1: LLM returns tool call for "weather" + round1TC := []llm.ToolCall{tc("call-1", "weather", map[string]any{"city": "Beijing"})} + // Round 2: LLM returns final answer with no tool calls + round2Done := doneEv(nil) + round2Done.Delta = "The weather is sunny" + + mock := &multiSeqMock{ + rounds: []iter.Seq2[*llm.Event, error]{ + func(yield func(*llm.Event, error) bool) { + yield(deltaEv("Let me check"), nil) + yield(doneEv(round1TC), nil) + }, + func(yield func(*llm.Event, error) bool) { + yield(round2Done, nil) + }, + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(reg), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + events, err := collectEvents(loop.Run(context.Background(), nil, nil)) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Should have: round 1 events (2) + tool result event (1) + round 2 events (1) = 4 + if len(events) < 4 { + t.Fatalf("expected at least 4 events, got %d", len(events)) + } + + // Find tool result event + var foundToolResult bool + for _, ev := range events { + if ev.ToolResult != nil { + foundToolResult = true + if ev.ToolResult.Name != "weather" { + t.Errorf("tool result name = %q, want weather", ev.ToolResult.Name) + } + if ev.ToolResult.Content != "sunny, 22°C" { + t.Errorf("tool result content = %q", ev.ToolResult.Content) + } + break + } + } + if !foundToolResult { + t.Error("no tool result event found") + } +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 3 — non-streaming RunNonStreaming +// --------------------------------------------------------------------------- + +func TestAgentLoopRunNonStreaming(t *testing.T) { + mock := &chatMock{ + result: &llm.ChatResult{ + Content: "The capital of France is Paris.", + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(testRegistry()), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + answer, err := loop.RunNonStreaming(context.Background(), nil, nil) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if answer != "The capital of France is Paris." { + t.Errorf("answer = %q, want 'The capital of France is Paris.'", answer) + } +} + +func TestAgentLoopRunNonStreaming_WithTools(t *testing.T) { + reg := testRegistry() + reg.RegisterInvoker("calc", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return echoResult("42"), nil + }) + + tc1 := []llm.ToolCall{tc("c1", "calc", map[string]any{"expr": "6*7"})} + + mock := &multiChatMock{ + results: []*llm.ChatResult{ + {ToolCalls: tc1, Thinking: "need to calculate"}, + {Content: "The answer is 42."}, + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(reg), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + answer, err := loop.RunNonStreaming(context.Background(), nil, nil) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if answer != "The answer is 42." { + t.Errorf("answer = %q, want 'The answer is 42.'", answer) + } +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 4 — maxLoop limit +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_MaxLoopExceeded(t *testing.T) { + // LLM that always returns a tool call, forcing infinite loop. + // With MaxLoop=2, it should stop after 2 iterations with an error. + reg := testRegistry() + var invokeCount int32 + reg.RegisterInvoker("dummy", func(ctx context.Context, params map[string]any) (map[string]any, error) { + atomic.AddInt32(&invokeCount, 1) + return echoResult("done"), nil + }) + + tcDummy := []llm.ToolCall{tc("c1", "dummy", nil)} + + mock := &multiSeqMock{ + rounds: []iter.Seq2[*llm.Event, error]{ + func(yield func(*llm.Event, error) bool) { + yield(deltaEv("calling tool"), nil) + yield(doneEv(tcDummy), nil) + }, + func(yield func(*llm.Event, error) bool) { + yield(deltaEv("calling tool again"), nil) + yield(doneEv(tcDummy), nil) + }, + func(yield func(*llm.Event, error) bool) { + yield(deltaEv("third call"), nil) + yield(doneEv(tcDummy), nil) + }, + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(reg), + MaxLoop: 2, + } + loop := NewAgentLoop(cfg) + + events, err := collectEvents(loop.Run(context.Background(), nil, nil)) + + if err == nil { + t.Fatal("expected an error for maxLoop exceeded, got nil") + } + + t.Logf("maxLoop error: %v", err) + t.Logf("events collected: %d", len(events)) + t.Logf("tool invocations: %d", invokeCount) + + // Should have stopped: events from round 1 (2) + tool result (1) + // + round 2 (2) + tool result (1) = 6, then maxLoop triggered before round 3 + if len(events) < 4 { + t.Errorf("expected at least 4 events before maxLoop error, got %d", len(events)) + } + if atomic.LoadInt32(&invokeCount) != 2 { + t.Errorf("expected 2 tool invocations, got %d", invokeCount) + } +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 5 — tool execution allTerminate=true +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_AllTerminate(t *testing.T) { + reg := testRegistry() + reg.RegisterInvoker("finish", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return echoTerminateResult("all done"), nil + }) + + tcFinish := []llm.ToolCall{tc("c1", "finish", nil)} + + // Even though round 2 exists, it should never be called because allTerminate=true + mock := &multiSeqMock{ + rounds: []iter.Seq2[*llm.Event, error]{ + func(yield func(*llm.Event, error) bool) { + yield(deltaEv("finishing"), nil) + yield(doneEv(tcFinish), nil) + }, + func(yield func(*llm.Event, error) bool) { + // This should NOT be reached + yield(deltaEv("SHOULD NOT APPEAR"), nil) + yield(doneEv(nil), nil) + }, + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(reg), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + events, err := collectEvents(loop.Run(context.Background(), nil, nil)) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Verify we got the tool result with terminate + var foundTerminate bool + for _, ev := range events { + if ev.ToolResult != nil && ev.ToolResult.Terminate { + foundTerminate = true + } + if ev.Delta == "SHOULD NOT APPEAR" { + t.Error("second LLM call should not have happened after allTerminate") + } + } + if !foundTerminate { + t.Error("tool result with terminate=true not found") + } +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 6 — context cancel +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_ContextCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + // Mock that yields a few events then blocks until context is cancelled + mock := &singleSeqMock{seq: func(yield func(*llm.Event, error) bool) { + yield(deltaEv("processing"), nil) + // Cancel context mid-stream + cancel() + // Try to yield another event — simulation of cancellation + // The StreamChat implementation would check ctx.Done() + select { + case <-ctx.Done(): + yield(nil, ctx.Err()) + case <-time.After(100 * time.Millisecond): + yield(doneEv(nil), nil) + } + }} + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(testRegistry()), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + events, err := collectEvents(loop.Run(ctx, nil, nil)) + + if err == nil { + // Could be that context cancel didn't propagate in mock + t.Log("context cancel did not produce error (mock timing dependent)") + } + _ = events + // At minimum, iteration terminated without hanging + t.Logf("context cancel test: %d events, err=%v", len(events), err) +} + +// --------------------------------------------------------------------------- +// Tests: Scenario 7 — StreamChat error +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_StreamChatError(t *testing.T) { + streamErr := errors.New("API connection refused") + mock := &singleSeqMock{seq: func(yield func(*llm.Event, error) bool) { + yield(deltaEv("partial response"), nil) + yield(nil, streamErr) + }} + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(testRegistry()), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + events, err := collectEvents(loop.Run(context.Background(), nil, nil)) + + if err == nil { + t.Fatal("expected error from StreamChat, got nil") + } + if !errors.Is(err, streamErr) { + t.Logf("error chain: %v (contains streamErr=%v)", err, errors.Is(err, streamErr)) + } + if len(events) < 1 { + t.Error("expected at least the partial response event before error") + } +} + +// --------------------------------------------------------------------------- +// Tests: Option pattern +// --------------------------------------------------------------------------- + +func TestAgentLoop_WithMaxLoop(t *testing.T) { + cfg := AgentLoopConfig{ + MaxLoop: 0, + } + loop := NewAgentLoop(cfg, WithMaxLoop(10)) + if loop.cfg.MaxLoop != 10 { + t.Errorf("WithMaxLoop: expected 10, got %d", loop.cfg.MaxLoop) + } +} + +func TestAgentLoop_DefaultMaxLoop(t *testing.T) { + cfg := AgentLoopConfig{ + MaxLoop: 0, + } + loop := NewAgentLoop(cfg) + if loop.cfg.MaxLoop != 5 { + t.Errorf("default MaxLoop: expected 5, got %d", loop.cfg.MaxLoop) + } +} + +// --------------------------------------------------------------------------- +// Tests: Run closure handles early termination by caller +// --------------------------------------------------------------------------- + +func TestAgentLoopRun_CallerBreaksEarly(t *testing.T) { + // Simulate caller breaking out of range early — iterator should stop + mock := &singleSeqMock{seq: func(yield func(*llm.Event, error) bool) { + if !yield(deltaEv("event-1"), nil) { + return + } + // This should not be reached if caller breaks + panic("should not yield event-2 if caller broke early") + }} + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(testRegistry()), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + var count int + for ev, err := range loop.Run(context.Background(), nil, nil) { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + count++ + if count >= 1 { + break // early termination + } + _ = ev + } + if count != 1 { + t.Errorf("expected 1 event before break, got %d", count) + } +} + +// --------------------------------------------------------------------------- +// Tests: RunNonStreaming with MaxLoop exceeded +// --------------------------------------------------------------------------- + +func TestAgentLoopRunNonStreaming_MaxLoopExceeded(t *testing.T) { + reg := testRegistry() + reg.RegisterInvoker("dummy", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return echoResult("ok"), nil + }) + + tcDummy := []llm.ToolCall{tc("c1", "dummy", nil)} + + // Always returns tool calls — should hit maxLoop + mock := &multiChatMock{ + results: []*llm.ChatResult{ + {ToolCalls: tcDummy}, + {ToolCalls: tcDummy}, + {ToolCalls: tcDummy}, + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(reg), + MaxLoop: 2, + } + loop := NewAgentLoop(cfg) + + answer, err := loop.RunNonStreaming(context.Background(), nil, nil) + + if err == nil { + t.Fatal("expected error for maxLoop exceeded in non-streaming mode, got nil") + } + t.Logf("non-streaming maxLoop error: %v, answer: %q", err, answer) +} + +// --------------------------------------------------------------------------- +// Tests: RunNonStreaming with allTerminate +// --------------------------------------------------------------------------- + +func TestAgentLoopRunNonStreaming_AllTerminate(t *testing.T) { + reg := testRegistry() + reg.RegisterInvoker("finish", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return echoTerminateResult("done"), nil + }) + + tcFinish := []llm.ToolCall{tc("c1", "finish", nil)} + + mock := &multiChatMock{ + results: []*llm.ChatResult{ + {ToolCalls: tcFinish}, + {Content: "should not reach"}, // should NOT be consumed + }, + } + + cfg := AgentLoopConfig{ + LLM: mock, + ToolExec: NewToolExecutor(reg), + MaxLoop: 5, + } + loop := NewAgentLoop(cfg) + + answer, err := loop.RunNonStreaming(context.Background(), nil, nil) + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + // When allTerminate, we return the content from the first Chat (which had tool calls), + // so answer may be empty (tool calls only). This verifies no error and no loop. + t.Logf("non-streaming allTerminate answer: %q", answer) + // The key assertion: mock.idx should be 1 (only first result consumed) + if mock.idx != 1 { + t.Errorf("expected 1 Chat call, got %d (should have stopped at allTerminate)", mock.idx) + } +} diff --git a/pkg/services/agent/tool_executor.go b/pkg/services/agent/tool_executor.go new file mode 100644 index 0000000..42b99f1 --- /dev/null +++ b/pkg/services/agent/tool_executor.go @@ -0,0 +1,229 @@ +package agent + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + "time" + + "github.com/liut/morign/pkg/services/llm" + "github.com/liut/morign/pkg/services/tools" +) + +// ChatExecutor 定义聊天执行函数类型，支持流式/非流式 +type ChatExecutor func(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, []llm.ToolCall, *llm.Usage, error) + +// ToolExecutor 封装工具调用循环逻辑 +type ToolExecutor struct { + toolreg *tools.Registry + + // BeforeToolCall 在每个工具调用前执行。返回 block=true 时阻止工具执行， + // reason 作为 error 内容写入 ToolResult。可选，nil 时跳过。 + BeforeToolCall func(ctx context.Context, name string, params map[string]any) (block bool, reason string) + + // AfterToolCall 在每个工具成功执行后调用，可修改 result。 + // 返回的 map 会替代原 result 用于后续格式化。可选，nil 时跳过。 + AfterToolCall func(ctx context.Context, name string, result map[string]any) map[string]any +} + +// NewToolExecutor 创建 ToolExecutor +func NewToolExecutor(toolreg *tools.Registry) *ToolExecutor { + return &ToolExecutor{toolreg: toolreg} +} + +// ExecuteToolCallLoop 执行工具调用循环，直到无 tool calls +func (e *ToolExecutor) ExecuteToolCallLoop( + ctx context.Context, + messages []llm.Message, + tools []llm.ToolDefinition, + exec ChatExecutor, +) (string, []llm.ToolCall, *llm.Usage, error) { + for { + answer, toolCalls, usage, err := exec(ctx, messages, tools) + if err != nil { + return "", nil, nil, err + } + + if len(toolCalls) == 0 { + return answer, nil, usage, nil + } + + evs, msgs, allTerminate := e.ExecuteToolCalls(ctx, messages, toolCalls, "") + messages = msgs + if allTerminate { + return answer, toolCalls, usage, nil + } + if len(evs) == 0 { + // 没有成功执行任何工具，跳出循环 + return answer, toolCalls, usage, nil + } + } +} + +// ExecuteToolCalls 执行单轮工具调用（并发），返回事件列表、更新后的消息和 allTerminate。 +// allTerminate 在本次所有成功执行的 tool 都标记 terminate 时为 true。 +func (e *ToolExecutor) ExecuteToolCalls(ctx context.Context, messages []llm.Message, toolCalls []llm.ToolCall, think string) ([]*llm.Event, []llm.Message, bool) { + if len(toolCalls) == 0 { + return nil, messages, false + } + + messages = append(messages, llm.Message{ + Role: llm.RoleAssistant, + Thinking: think, + ToolCalls: toolCalls, + }) + + n := len(toolCalls) + results := make([]*llm.Event, n) + var wg sync.WaitGroup + wg.Add(n) + + for i, tc := range toolCalls { + go func(idx int, tc llm.ToolCall) { + defer wg.Done() + defer func() { + if r := recover(); r != nil { + slog.Error("tool call panic recovered", "tool", tc.Function.Name, "panic", r) + } + }() + + slog.Info("chat", "toolCallID", tc.ID, "toolCallType", tc.Type, "toolCallName", tc.Function.Name) + + if tc.Type != "function" { + return + } + + var parameters map[string]any + args := string(tc.Function.Arguments) + if args != "" && args != "{}" { + if err := json.Unmarshal(tc.Function.Arguments, ¶meters); err != nil { + slog.Info("chat", "toolCallID", tc.ID, "args", args, "err", err) + return + } + } + if parameters == nil { + parameters = make(map[string]any) + } + + // Before hook + if e.BeforeToolCall != nil { + block, reason := e.BeforeToolCall(ctx, tc.Function.Name, parameters) + if block { + results[idx] = &llm.Event{ + ID: llm.NewEventID(), + Timestamp: time.Now(), + Author: tc.Function.Name, + ToolResult: &llm.ToolResult{ + CallID: tc.ID, + Name: tc.Function.Name, + Content: reason, + }, + } + return + } + } + + content, err := e.toolreg.Invoke(ctx, tc.Function.Name, parameters) + if err != nil { + slog.Info("invokeTool fail", "toolCallName", tc.Function.Name, "err", err) + return + } + + slog.Info("invokeTool ok", "toolCallName", tc.Function.Name, + "content", tools.ResultLogs(content)) + + // After hook + if e.AfterToolCall != nil { + content = e.AfterToolCall(ctx, tc.Function.Name, content) + } + + // Extract terminate from result + terminate := false + if t, ok := content["terminate"]; ok { + if tb, ok := t.(bool); ok { + terminate = tb + } + } + + toolResult := formatToolResult(content) + results[idx] = &llm.Event{ + ID: llm.NewEventID(), + Timestamp: time.Now(), + Author: tc.Function.Name, + ToolResult: &llm.ToolResult{ + CallID: tc.ID, + Name: tc.Function.Name, + Content: toolResult, + Terminate: terminate, + }, + } + }(i, tc) + } + + wg.Wait() + + // Collect results in call order, build messages + var events []*llm.Event + allTerminate := true + hasResult := false + for i, ev := range results { + if ev == nil { + allTerminate = false + continue + } + hasResult = true + if !ev.ToolResult.Terminate { + allTerminate = false + } + events = append(events, ev) + messages = append(messages, llm.Message{ + Role: llm.RoleTool, + Content: ev.ToolResult.Content, + ToolCallID: toolCalls[i].ID, + }) + } + + if !hasResult { + allTerminate = false + } + + return events, messages, allTerminate +} + +// formatToolResult 将工具结果转换为文本字符串 +// 优先提取 content 数组中的 text，否则使用 structuredContent +func formatToolResult(result map[string]any) string { + if result == nil { + return "" + } + // 优先提取 content 数组中的 text + if content, ok := result["content"].([]any); ok { + for _, c := range content { + if cMap, ok := c.(map[string]any); ok { + if text, ok := cMap["text"].(string); ok && text != "" { + return text + } + } + } + } + // 备选：使用 structuredContent + if sc, ok := result["structuredContent"].(string); ok { + return sc + } + if sc, ok := result["structuredContent"].(map[string]any); ok { + for k, v := range sc { + if s, ok := v.(string); ok && k == "text" { + return s + } + } + if b, err := json.Marshal(sc); err == nil { + return string(b) + } + } + // 最后：序列化为 JSON + if b, err := json.Marshal(result); err == nil { + return string(b) + } + return "" +} diff --git a/pkg/services/agent/tool_executor_test.go b/pkg/services/agent/tool_executor_test.go new file mode 100644 index 0000000..4f5362c --- /dev/null +++ b/pkg/services/agent/tool_executor_test.go @@ -0,0 +1,300 @@ +package agent + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + + "github.com/liut/morign/pkg/services/llm" + "github.com/liut/morign/pkg/services/tools" +) + +func testRegistry() *tools.Registry { + return tools.NewRegistry(nil) +} + +func tc(id, name string, args map[string]any) llm.ToolCall { + raw, _ := json.Marshal(args) + return llm.ToolCall{ + ID: id, + Type: "function", + Function: llm.ToolCallFunc{ + Name: name, + Arguments: raw, + }, + } +} + +func TestExecuteToolCallsConcurrent(t *testing.T) { + // Scenario 1: Happy path — two tools execute concurrently, results in call order. + reg := testRegistry() + var orderMu sync.Mutex + var order []string + ready := make(chan struct{}) + + reg.RegisterInvoker("alpha", func(ctx context.Context, params map[string]any) (map[string]any, error) { + orderMu.Lock() + order = append(order, "alpha") + orderMu.Unlock() + ready <- struct{}{} // signal started + return map[string]any{"content": []any{map[string]any{"text": "alpha-result"}}}, nil + }) + reg.RegisterInvoker("beta", func(ctx context.Context, params map[string]any) (map[string]any, error) { + orderMu.Lock() + order = append(order, "beta") + orderMu.Unlock() + <-ready // wait for alpha to also start (proves concurrency) + return map[string]any{"content": []any{map[string]any{"text": "beta-result"}}}, nil + }) + + te := NewToolExecutor(reg) + calls := []llm.ToolCall{ + tc("c1", "alpha", nil), + tc("c2", "beta", nil), + } + + evs, msgs, allTerm := te.ExecuteToolCalls(context.Background(), nil, calls, "") + + if len(order) != 2 { + t.Errorf("expected 2 invocations, got %d", len(order)) + } + if len(evs) != 2 { + t.Fatalf("expected 2 events, got %d", len(evs)) + } + if evs[0].ToolResult.CallID != "c1" { + t.Errorf("first event CallID = %q, want c1", evs[0].ToolResult.CallID) + } + if evs[1].ToolResult.CallID != "c2" { + t.Errorf("second event CallID = %q, want c2", evs[1].ToolResult.CallID) + } + if len(msgs) != 3 { + t.Fatalf("expected 3 messages (nil+assistant+2 tool results), got %d", len(msgs)) + } + if msgs[0].Role != llm.RoleAssistant { + t.Errorf("first msg role = %q, want assistant", msgs[0].Role) + } + if allTerm { + t.Error("allTerminate should be false when no tool sets terminate") + } +} + +func TestExecuteToolCallsBeforeHookAllow(t *testing.T) { + // Scenario 2: BeforeToolCall returns block=false, tool executes normally. + reg := testRegistry() + reg.RegisterInvoker("echo", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "ok"}}}, nil + }) + + te := NewToolExecutor(reg) + var beforeCalled bool + te.BeforeToolCall = func(ctx context.Context, name string, params map[string]any) (bool, string) { + beforeCalled = true + return false, "" + } + + evs, _, _ := te.ExecuteToolCalls(context.Background(), nil, + []llm.ToolCall{tc("c1", "echo", nil)}, "") + + if !beforeCalled { + t.Error("BeforeToolCall was not invoked") + } + if len(evs) != 1 { + t.Fatalf("expected 1 event, got %d", len(evs)) + } + if evs[0].ToolResult.Content != "ok" { + t.Errorf("result = %q, want ok", evs[0].ToolResult.Content) + } +} + +func TestExecuteToolCallsBeforeHookBlock(t *testing.T) { + // Scenario 4: BeforeToolCall returns block=true, tool does not execute. + reg := testRegistry() + var invoked bool + reg.RegisterInvoker("echo", func(ctx context.Context, params map[string]any) (map[string]any, error) { + invoked = true + return map[string]any{"content": []any{map[string]any{"text": "ok"}}}, nil + }) + + te := NewToolExecutor(reg) + te.BeforeToolCall = func(ctx context.Context, name string, params map[string]any) (bool, string) { + return true, "blocked by policy" + } + + evs, _, _ := te.ExecuteToolCalls(context.Background(), nil, + []llm.ToolCall{tc("c1", "echo", nil)}, "") + + if invoked { + t.Error("tool should NOT have been invoked when blocked") + } + if len(evs) != 1 { + t.Fatalf("expected 1 event, got %d", len(evs)) + } + if evs[0].ToolResult.Content != "blocked by policy" { + t.Errorf("error content = %q, want 'blocked by policy'", evs[0].ToolResult.Content) + } + if evs[0].ToolResult.Terminate { + t.Error("blocked tool should NOT set Terminate=true by default") + } +} + +func TestExecuteToolCallsAfterHook(t *testing.T) { + // Scenario 3: AfterToolCall modifies the result. + reg := testRegistry() + reg.RegisterInvoker("echo", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "original"}}}, nil + }) + + te := NewToolExecutor(reg) + te.AfterToolCall = func(ctx context.Context, name string, result map[string]any) map[string]any { + result["content"] = []any{map[string]any{"text": "modified"}} + return result + } + + evs, _, _ := te.ExecuteToolCalls(context.Background(), nil, + []llm.ToolCall{tc("c1", "echo", nil)}, "") + + if len(evs) != 1 { + t.Fatalf("expected 1 event, got %d", len(evs)) + } + if evs[0].ToolResult.Content != "modified" { + t.Errorf("after hook should modify content, got %q", evs[0].ToolResult.Content) + } +} + +func TestExecuteToolCallsAllTerminate(t *testing.T) { + // Scenario 5: All tools return terminate=true → allTerminate=true. + reg := testRegistry() + reg.RegisterInvoker("a", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "done"}}, "terminate": true}, nil + }) + reg.RegisterInvoker("b", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "done"}}, "terminate": true}, nil + }) + + te := NewToolExecutor(reg) + _, _, allTerm := te.ExecuteToolCalls(context.Background(), nil, + []llm.ToolCall{tc("c1", "a", nil), tc("c2", "b", nil)}, "") + + if !allTerm { + t.Error("allTerminate should be true when all tools set terminate") + } +} + +func TestExecuteToolCallsPartialTerminate(t *testing.T) { + // Scenario 6: Only some tools terminate → allTerminate=false. + reg := testRegistry() + reg.RegisterInvoker("a", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "t"}}, "terminate": true}, nil + }) + reg.RegisterInvoker("b", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "nt"}}}, nil + }) + + te := NewToolExecutor(reg) + _, _, allTerm := te.ExecuteToolCalls(context.Background(), nil, + []llm.ToolCall{tc("c1", "a", nil), tc("c2", "b", nil)}, "") + + if allTerm { + t.Error("allTerminate should be false when not all tools terminate") + } +} + +func TestExecuteToolCallsEmpty(t *testing.T) { + // Scenario 7: No tool calls → empty results. + te := NewToolExecutor(testRegistry()) + evs, msgs, allTerm := te.ExecuteToolCalls(context.Background(), nil, nil, "") + + if len(evs) != 0 { + t.Errorf("expected 0 events, got %d", len(evs)) + } + if len(msgs) != 0 { + t.Errorf("expected 0 messages (nil in), got %d", len(msgs)) + } + if allTerm { + t.Error("allTerminate should be false for empty calls") + } +} + +func TestExecuteToolCallsInvokeError(t *testing.T) { + // Tool invocation error: tool call returns error → no event for that tool. + reg := testRegistry() + reg.RegisterInvoker("failer", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return nil, errors.New("boom") + }) + reg.RegisterInvoker("ok", func(ctx context.Context, params map[string]any) (map[string]any, error) { + return map[string]any{"content": []any{map[string]any{"text": "ok"}}}, nil + }) + + te := NewToolExecutor(reg) + evs, _, _ := te.ExecuteToolCalls(context.Background(), nil, + []llm.ToolCall{tc("c1", "failer", nil), tc("c2", "ok", nil)}, "") + + // Only "ok" succeeds; failer produces no event + if len(evs) != 1 { + t.Fatalf("expected 1 event, got %d", len(evs)) + } + if evs[0].ToolResult.CallID != "c2" { + t.Errorf("event CallID = %q, want c2", evs[0].ToolResult.CallID) + } +} + +func TestExecuteToolCallsNonFunctionType(t *testing.T) { + // Non-"function" type tool calls are skipped. + reg := testRegistry() + var invoked bool + reg.RegisterInvoker("echo", func(ctx context.Context, params map[string]any) (map[string]any, error) { + invoked = true + return map[string]any{"content": []any{map[string]any{"text": "ok"}}}, nil + }) + + te := NewToolExecutor(reg) + calls := []llm.ToolCall{ + {ID: "c1", Type: "retrieval", Function: llm.ToolCallFunc{Name: "echo", Arguments: []byte("{}")}}, + } + evs, _, _ := te.ExecuteToolCalls(context.Background(), nil, calls, "") + + if invoked { + t.Error("non-function tool should be skipped") + } + if len(evs) != 0 { + t.Errorf("expected 0 events, got %d", len(evs)) + } +} +// moved from pkg/web/api/handle_convo_test.go +func TestFormatToolResult(t *testing.T) { + tests := []struct { + name string + input map[string]any + expected string + }{ + { + name: "nil input", + input: nil, + expected: "", + }, + { + name: "empty map", + input: map[string]any{}, + expected: "{}", + }, + { + name: "normal map", + input: map[string]any{ + "result": "success", + "count": 1, + }, + expected: `{"count":1,"result":"success"}`, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := formatToolResult(tt.input) + if result != tt.expected { + t.Errorf("formatToolResult() = %v, want %v", result, tt.expected) + } + }) + } +} diff --git a/pkg/services/llm/event.go b/pkg/services/llm/event.go index 9a6e16a..d949cc2 100644 --- a/pkg/services/llm/event.go +++ b/pkg/services/llm/event.go @@ -46,9 +46,10 @@ type Event struct { // ToolResult 携带单次工具调用的执行结果。 type ToolResult struct { - CallID string - Name string - Content string + CallID string + Name string + Content string + Terminate bool // 工具请求终止 agent 循环 } // EventActions 是 Event 携带的副作用指令。 diff --git a/pkg/services/tools/registry.go b/pkg/services/tools/registry.go index fe97306..db3503a 100644 --- a/pkg/services/tools/registry.go +++ b/pkg/services/tools/registry.go @@ -81,6 +81,17 @@ func NewRegistry(sto stores.Storage, opts ...RegistryOption) *Registry { return r } +// RegisterInvoker 注册一个工具及其调用函数。name 为空或 inv 为 nil 时静默忽略。 +func (r *Registry) RegisterInvoker(name string, inv Invoker) { + if name == "" || inv == nil { + return + } + r.toolsMu.Lock() + defer r.toolsMu.Unlock() + r.tools = append(r.tools, mcps.ToolDescriptor{Name: name}) + r.invokers[name] = inv +} + // Invoke 调用指定名称的工具，频道工具优先 func (r *Registry) Invoke(ctx context.Context, name string, params map[string]any) (map[string]any, error) { if name == "" { diff --git a/pkg/web/api/agent.go b/pkg/web/api/agent.go index df1742a..3e5a9e4 100644 --- a/pkg/web/api/agent.go +++ b/pkg/web/api/agent.go @@ -6,6 +6,7 @@ import ( "iter" "strings" + "github.com/liut/morign/pkg/services/agent" "github.com/liut/morign/pkg/services/llm" "github.com/liut/morign/pkg/services/tools" "github.com/liut/morign/pkg/settings" @@ -21,7 +22,7 @@ type StreamCallbacks struct { type Agent struct { llm llm.Client toolreg *tools.Registry - toolExec *ToolExecutor + toolExec *agent.ToolExecutor sysPrompt string toolsPrompt string } @@ -31,7 +32,7 @@ func NewAgent(llmClient llm.Client, toolreg *tools.Registry, sysPrompt, toolsPro return &Agent{ llm: llmClient, toolreg: toolreg, - toolExec: NewToolExecutor(toolreg), + toolExec: agent.NewToolExecutor(toolreg), sysPrompt: sysPrompt, toolsPrompt: toolsPrompt, } @@ -65,66 +66,28 @@ func (ag *Agent) BuildSystemMessage(ctx context.Context) (llm.Message, []llm.Too return llm.Message{Role: llm.RoleSystem, Content: sb.String()}, tools } -// Chat 非流式对话，直接使用 llm.Chat + 工具调用循环。 +// Chat 非流式对话，使用 AgentLoop.RunNonStreaming。 func (ag *Agent) Chat(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, error) { - exec := func(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, []llm.ToolCall, *llm.Usage, error) { - result, err := ag.llm.Chat(ctx, messages, tools) - if err != nil { - return "", nil, nil, err - } - return result.Content, result.ToolCalls, result.Usage, nil - } - answer, _, _, err := ag.toolExec.ExecuteToolCallLoop(ctx, messages, tools, exec) - return answer, err + loop := agent.NewAgentLoop(agent.AgentLoopConfig{ + LLM: ag.llm, + ToolExec: ag.toolExec, + }, agent.WithMaxLoop(settings.Current.MaxLoopIterations)) + + return loop.RunNonStreaming(ctx, messages, tools) } -// Run 以 iter.Seq2 方式执行对话，包含工具调用循环。 +// Run 以 iter.Seq2 方式执行对话，使用 AgentLoop。 func (ag *Agent) Run(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) iter.Seq2[*llm.Event, error] { - maxLoop := settings.Current.MaxLoopIterations - if maxLoop <= 0 { - maxLoop = 5 - } + loop := agent.NewAgentLoop(agent.AgentLoopConfig{ + LLM: ag.llm, + ToolExec: ag.toolExec, + }, agent.WithMaxLoop(settings.Current.MaxLoopIterations)) return func(yield func(*llm.Event, error) bool) { - var fullThink string - - for iter := 0; iter < maxLoop; iter++ { - var roundAnswer string - var roundThink string - var toolCalls []llm.ToolCall - - for event, err := range ag.llm.StreamChat(ctx, messages, tools) { - if err != nil { - yield(nil, fmt.Errorf("stream chat: %w", err)) - return - } - roundAnswer += event.Delta - roundThink += event.Think - - if !yield(event, nil) { - return - } - - if event.Done { - toolCalls = event.ToolCalls - } - } - - fullThink += roundThink - - if len(toolCalls) == 0 { + for event, err := range loop.Run(ctx, messages, tools) { + if !yield(event, err) { return } - - // 执行工具调用，产生 tool result events - events, updatedMsgs := ag.toolExec.ExecuteToolCalls(ctx, messages, toolCalls, roundThink) - messages = updatedMsgs - for _, ev := range events { - ev.Author = "tool" - if !yield(ev, nil) { - return - } - } } } } diff --git a/pkg/web/api/api.go b/pkg/web/api/api.go index 95774a0..0ae6eeb 100644 --- a/pkg/web/api/api.go +++ b/pkg/web/api/api.go @@ -17,6 +17,7 @@ import ( "github.com/liut/morign/pkg/models/aigc" "github.com/liut/morign/pkg/models/mcps" + "github.com/liut/morign/pkg/services/agent" "github.com/liut/morign/pkg/services/llm" "github.com/liut/morign/pkg/services/runner" "github.com/liut/morign/pkg/services/stores" @@ -51,7 +52,7 @@ type api struct { llm llm.Client preset aigc.Preset toolreg *tools.Registry - toolExec *ToolExecutor + toolExec *agent.ToolExecutor rnr *runner.Runner } @@ -109,7 +110,7 @@ func newapi(sto stores.Storage) *api { llm: llmClient, preset: preset, toolreg: toolreg, - toolExec: NewToolExecutor(toolreg), + toolExec: agent.NewToolExecutor(toolreg), rnr: runner.New(stores.NewSessionStore(sto), stores.NewHistoryStore(sto)), } } diff --git a/pkg/web/api/convo_basic.go b/pkg/web/api/convo_basic.go index 4a4c035..f80eeb5 100644 --- a/pkg/web/api/convo_basic.go +++ b/pkg/web/api/convo_basic.go @@ -2,9 +2,6 @@ package api import ( "time" - - "github.com/liut/morign/pkg/services/llm" - "github.com/liut/morign/pkg/utils/words" ) const ( @@ -99,18 +96,4 @@ type ChatMessage struct { Title string `json:"title,omitempty"` } -// wrap response from llm -type chatResponse struct { - answer string - toolCalls []llm.ToolCall - usage *llm.Usage - finish llm.FinishReason - think string // reasoning_content for DeepSeek thinking mode - - model string - llmResID string -} -func cutTxt(s string, n int, opts ...string) string { - return words.TakeHead(s, n, opts...) -} diff --git a/pkg/web/api/handle_convo.go b/pkg/web/api/handle_convo.go index 7767eed..215b36b 100644 --- a/pkg/web/api/handle_convo.go +++ b/pkg/web/api/handle_convo.go @@ -19,6 +19,7 @@ import ( "github.com/liut/morign/pkg/models/convo" "github.com/liut/morign/pkg/models/corpus" "github.com/liut/morign/pkg/models/mcps" + "github.com/liut/morign/pkg/services/agent" "github.com/liut/morign/pkg/services/llm" "github.com/liut/morign/pkg/services/stores" "github.com/liut/morign/pkg/services/tools" @@ -54,8 +55,6 @@ type chatRequest struct { tools []llm.ToolDefinition isSSE bool cs stores.Conversation - hi *aigc.HistoryItem - chunkIdx int // 全局 chunk 计数器，用于 SSE 事件序号 prompt string } @@ -202,13 +201,7 @@ func (a *api) prepareChatRequest(ctx context.Context, param *ChatRequest) *chatR messages: messages, tools: tools, cs: cs, - hi: &aigc.HistoryItem{ - Time: time.Now().Unix(), - ChatItem: &aigc.HistoryChatItem{ - User: param.Prompt, - }, - }, - prompt: param.Prompt, + prompt: param.Prompt, } } @@ -231,47 +224,22 @@ func (a *api) postChat(w http.ResponseWriter, r *http.Request) { isSSE := param.Stream || strings.HasSuffix(r.URL.Path, "-sse") isStream := param.Stream || isSSE ccr := a.prepareChatRequest(r.Context(), ¶m) - ccr.isSSE = isSSE + loop := agent.NewAgentLoop(agent.AgentLoopConfig{ + LLM: a.llm, + ToolExec: a.toolExec, + MaxLoop: settings.Current.MaxLoopIterations, + }) + logger().Infow("chat", "csid", param.GetConversionID(), "msgs", len(ccr.messages), "prompt", param.Prompt, "ip", r.RemoteAddr) if isStream { - res := a.chatStreamResponseLoop(ccr, w, r) - logger().Infow("stream response", "answer_len", len(res.answer), "toolCalls_len", len(res.toolCalls)) - if len(res.answer) > 0 { - - // TODO: migrate to convo.Message - if settings.Current.QAChatLog { - in := corpus.ChatLogBasic{ - ChatID: ccr.cs.GetOID(), - Question: param.Prompt, - Answer: res.answer, - } - ip, _, _ := strings.Cut(r.RemoteAddr, ":") - in.MetaAddKVs("ip", ip) - if res.usage != nil { - in.MetaAddKVs("usage", res.usage) - } - _, err := stores.Sgt().Corpus().CreateChatLog(r.Context(), in) - if err != nil { - logger().Infow("save chat log fail", "err", err) - } - } - } - + a.handleSSEStream(w, r, loop, ccr) return } - // 非流式场景：使用循环执行工具调用 - exec := func(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, []llm.ToolCall, *llm.Usage, error) { - result, err := a.llm.Chat(ctx, messages, tools) - if err != nil { - return "", nil, nil, err - } - return result.Content, result.ToolCalls, result.Usage, nil - } - answer, _, _, err := a.executeToolCallLoop(r.Context(), ccr.messages, ccr.tools, exec) + answer, err := loop.RunNonStreaming(r.Context(), ccr.messages, ccr.tools) if err != nil { apiFail(w, r, 500, err) return @@ -310,179 +278,146 @@ func writeEvent(w io.Writer, id string, m any) bool { return true } -// chatStreamResponseLoop 循环处理流式响应，支持工具调用循环 -func (a *api) chatStreamResponseLoop(ccr *chatRequest, w http.ResponseWriter, r *http.Request) (res chatResponse) { - // 预先设置 HTTP 头信息（只设置一次） +// handleSSEStream runs the agent loop and writes events to the SSE response. +func (a *api) handleSSEStream(w http.ResponseWriter, r *http.Request, loop *agent.AgentLoop, ccr *chatRequest) { if _, ok := w.(http.Flusher); !ok { http.Error(w, "Streaming unsupported!", http.StatusInternalServerError) - return chatResponse{} + return } w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.Header().Set("Content-Type", "text/event-stream") w.Header().Add("Conversation-ID", ccr.cs.GetID()) - w.(http.Flusher).Flush() - var iter int - maxLoopIterations := settings.Current.MaxLoopIterations - if maxLoopIterations <= 0 { - maxLoopIterations = 5 - } - cctx := stores.ContextWithConvoID(r.Context(), ccr.cs.GetID()) - for { - iter++ - // 达到迭代次数限制，跳出循环 - if iter > maxLoopIterations { - logger().Infow("chat loop iteration limit reached", "maxIter", maxLoopIterations) + var ( + answer string + think string + usage *llm.Usage + finish llm.FinishReason + chunkIdx int + lastWriteEmpty bool + ) + + for event, err := range loop.Run(cctx, ccr.messages, ccr.tools) { + if err != nil { + logger().Infow("agent loop error", "err", err) break } - // 调用流式响应处理 - streamRes := a.doChatStream(ccr, w, r) - logger().Infow("stream round done", "iter", iter, "maxIter", maxLoopIterations, - "answer_len", len(streamRes.answer), "toolCalls_len", len(streamRes.toolCalls)) - - // 累积答案 - res.answer += streamRes.answer - if streamRes.usage != nil { - res.usage = streamRes.usage + if event.ToolResult != nil { + continue } - // 如果没有工具调用，跳出循环 - if len(streamRes.toolCalls) == 0 { - res.finish = streamRes.finish - break + var cm ChatMessage + cm.Delta = event.Delta + cm.Think = event.Think + + answer += event.Delta + think += event.Think + + if event.Usage != nil { + usage = event.Usage } - logger().Infow("before execute tool calls", "tools", len(streamRes.toolCalls), "msgs", len(ccr.messages), - "think_len", len(streamRes.think)) - - // 执行工具调用，传入 reasoning_content 以便回传 - evs, msgs := a.toolExec.ExecuteToolCalls(cctx, ccr.messages, streamRes.toolCalls, streamRes.think) - ccr.messages = msgs - logger().Infow("executed tool calls", "executed", len(evs), "msgs", len(ccr.messages)) - if len(evs) == 0 { - // 没有成功执行任何工具，跳出循环 - res.finish = streamRes.finish - break + + if event.Done { + finish = event.StopReason + + if len(event.ToolCalls) > 0 && event.StopReason == llm.FinishReasonToolCalls { + cm.ToolCalls = convertToolCallsForJSON(event.ToolCalls) + chunkIdx++ + cm.ConversationID = ccr.cs.GetID() + cm.FinishReason = string(event.StopReason) + _ = writeEvent(w, strconv.Itoa(chunkIdx), &cm) + } + + if event.Usage != nil { + meta := map[string]any{ + "prompt": ccr.prompt, + "answerHead": words.TakeHead(answer, 12, ".."), + "answerTail": words.TakeTail(answer, 15, ".."), + } + if len(event.ResponseID) > 0 { + meta["resposeID"] = event.ResponseID + } + if err := a.rnr.Persist(r.Context(), ccr.cs.GetID(), &llm.Event{ + Author: "assistant", + Usage: event.Usage, + Model: event.Model, + MsgCount: len(ccr.messages), + Meta: meta, + }); err != nil { + logger().Infow("persist usage fail", "err", err) + } + } + + lastWriteEmpty = false + continue } + isEmpty := event.Delta == "" && event.Think == "" && len(cm.ToolCalls) == 0 + if !isEmpty || !lastWriteEmpty { + chunkIdx++ + if !writeEvent(w, strconv.Itoa(chunkIdx), &cm) { + break + } + } + lastWriteEmpty = isEmpty } - if len(res.answer) > 0 { + logger().Infow("stream response", "answer_len", len(answer), "finish", finish) + + if len(answer) > 0 { if err := a.rnr.Persist(r.Context(), ccr.cs.GetID(), &llm.Event{ Author: "assistant", - Delta: res.answer, - Think: res.think, + Delta: answer, + Think: think, UserPrompt: ccr.prompt, }); err != nil { logger().Infow("persist fail", "err", err) } } - // 请求完成处理：生成标题（限时同步执行） + if settings.Current.QAChatLog && len(answer) > 0 { + in := corpus.ChatLogBasic{ + ChatID: ccr.cs.GetOID(), + Question: ccr.prompt, + Answer: answer, + } + ip, _, _ := strings.Cut(r.RemoteAddr, ":") + in.MetaAddKVs("ip", ip) + if usage != nil { + in.MetaAddKVs("usage", usage) + } + _, err := stores.Sgt().Corpus().CreateChatLog(r.Context(), in) + if err != nil { + logger().Infow("save chat log fail", "err", err) + } + } + ctx, cancel := context.WithTimeout(r.Context(), 4*time.Second) defer cancel() - var cm ChatMessage - ccr.chunkIdx++ - cm.ConversationID = ccr.cs.GetID() - cm.FinishReason = string(res.finish) + var doneCm ChatMessage + chunkIdx++ + doneCm.ConversationID = ccr.cs.GetID() + doneCm.FinishReason = string(finish) history, err := ccr.cs.ListHistory(ctx) if err == nil && len(history) > 0 { title, err := stores.GetHistorySummary(ctx, history) if err == nil { - cm.Title = title + doneCm.Title = title } } - _ = writeEvent(w, strconv.Itoa(ccr.chunkIdx), &cm) - - // 发送完成事件（最后） - ccr.chunkIdx++ - _ = writeEvent(w, strconv.Itoa(ccr.chunkIdx), esDone) + _ = writeEvent(w, strconv.Itoa(chunkIdx), &doneCm) - return res -} - -// doChatStream 执行一次流式调用，返回累积的 answer 和 toolCalls -func (a *api) doChatStream(ccr *chatRequest, w http.ResponseWriter, r *http.Request) chatResponse { - var res chatResponse - var lastWriteEmpty bool // 标记上一次是否写入了空消息 - - for result, err := range a.llm.StreamChat(r.Context(), ccr.messages, ccr.tools) { - if err != nil { - logger().Infow("stream error", "err", err) - break - } - - var cm ChatMessage - - cm.Delta = result.Delta - cm.Think = result.Think - res.answer += result.Delta - res.think += result.Think - res.usage = result.Usage - if len(result.ToolCalls) > 0 && result.StopReason == llm.FinishReasonToolCalls { - cm.ToolCalls = convertToolCallsForJSON(result.ToolCalls) - ccr.chunkIdx++ - cm.ConversationID = ccr.cs.GetID() - cm.FinishReason = string(result.StopReason) - _ = writeEvent(w, strconv.Itoa(ccr.chunkIdx), &cm) - } - - if len(result.Model) > 0 { - res.model = result.Model - } - if len(result.ResponseID) > 0 { - res.llmResID = result.ResponseID - } - - if result.Done { - logger().Infow("result done", "finish", result.StopReason) - res.finish = result.StopReason - res.toolCalls = result.ToolCalls - break - } - - // 判断当前是否为空消息 - isEmpty := result.Delta == "" && result.Think == "" && len(cm.ToolCalls) == 0 - if !isEmpty || !lastWriteEmpty { - // 有内容，或者上一次不是空的，则输出 - ccr.chunkIdx++ - if wrote := writeEvent(w, strconv.Itoa(ccr.chunkIdx), &cm); !wrote { - break - } - } - // 如果当前是空的且上一次也是空的，跳过（连续空消息只保留第一个） - lastWriteEmpty = isEmpty - } - if res.usage != nil { - meta := map[string]any{ - "prompt": ccr.prompt, - "answerHead": words.TakeHead(res.answer, 12, ".."), - "answerTail": words.TakeTail(res.answer, 15, ".."), - } - if len(res.llmResID) > 0 { - meta["resposeID"] = res.llmResID - } - if err := a.rnr.Persist(r.Context(), ccr.cs.GetID(), &llm.Event{ - Author: "assistant", - Usage: res.usage, - Model: res.model, - MsgCount: len(ccr.messages), - Meta: meta, - }); err != nil { - logger().Infow("persist usage fail", "err", err) - } - } - logger().Infow("chat stream done", "finish", res.finish, "answer", len(res.answer), - "ahead", cutTxt(res.answer, 20)) - return res + chunkIdx++ + _ = writeEvent(w, strconv.Itoa(chunkIdx), esDone) } // @Tags 聊天 @@ -625,44 +560,6 @@ func (a *api) getTools(w http.ResponseWriter, r *http.Request) { apiOk(w, r, a.toolreg.ToolsFor(r.Context()), 0) } -// formatToolResult 将工具结果转换为文本字符串 -// 优先提取 content 数组中的 text，否则使用 structuredContent -func formatToolResult(result map[string]any) string { - if result == nil { - return "" - } - // logger().Debugw("formatToolResult", "result", result) - // 优先提取 content 数组中的 text - if content, ok := result["content"].([]any); ok { - for _, c := range content { - if cMap, ok := c.(map[string]any); ok { - if text, ok := cMap["text"].(string); ok && text != "" { - return text - } - } - } - } - // 备选：使用 structuredContent - if sc, ok := result["structuredContent"].(string); ok { - return sc - } - if sc, ok := result["structuredContent"].(map[string]any); ok { - for k, v := range sc { - if s, ok := v.(string); ok && k == "text" { - return s - } - } - if b, err := json.Marshal(sc); err == nil { - return string(b) - } - } - // 最后：序列化为 JSON - if b, err := json.Marshal(result); err == nil { - return string(b) - } - return "" -} - // convertToolCallsForJSON 将 llm.ToolCall 转换为可序列化的 map 格式 func convertToolCallsForJSON(tcs []llm.ToolCall) []map[string]any { if len(tcs) == 0 { @@ -685,12 +582,3 @@ func convertToolCallsForJSON(tcs []llm.ToolCall) []map[string]any { } return result } - -// executeToolCallLoop 执行工具调用循环，直到没有 tool calls -// - messages: 初始消息列表，会被修改 -// - tools: 工具定义 -// - exec: 执行聊天的函数（流式或非流式） -// 返回最终的 answer、最后的 toolCalls（如果有）、usage -func (a *api) executeToolCallLoop(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition, exec chatExecutor) (string, []llm.ToolCall, *llm.Usage, error) { - return a.toolExec.ExecuteToolCallLoop(ctx, messages, tools, exec) -} diff --git a/pkg/web/api/handle_convo_test.go b/pkg/web/api/handle_convo_test.go index 1cc1f77..005b6e5 100644 --- a/pkg/web/api/handle_convo_test.go +++ b/pkg/web/api/handle_convo_test.go @@ -7,41 +7,6 @@ import ( "github.com/liut/morign/pkg/services/llm" ) -func TestFormatToolResult(t *testing.T) { - tests := []struct { - name string - input map[string]any - expected string - }{ - { - name: "nil input", - input: nil, - expected: "", - }, - { - name: "empty map", - input: map[string]any{}, - expected: "{}", - }, - { - name: "normal map", - input: map[string]any{ - "result": "success", - "count": 1, - }, - expected: `{"count":1,"result":"success"}`, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - result := formatToolResult(tt.input) - if result != tt.expected { - t.Errorf("formatToolResult() = %v, want %v", result, tt.expected) - } - }) - } -} func TestConvertToolCallsForJSON(t *testing.T) { tests := []struct { diff --git a/pkg/web/api/handle_platform.go b/pkg/web/api/handle_platform.go index 368298a..01a64bb 100644 --- a/pkg/web/api/handle_platform.go +++ b/pkg/web/api/handle_platform.go @@ -11,6 +11,7 @@ import ( "github.com/liut/morign/pkg/models/aigc" "github.com/liut/morign/pkg/models/channel" "github.com/liut/morign/pkg/models/mcps" + "github.com/liut/morign/pkg/services/agent" "github.com/liut/morign/pkg/services/channels" "github.com/liut/morign/pkg/services/channels/feishu" "github.com/liut/morign/pkg/services/channels/wecom" @@ -26,7 +27,7 @@ type channelHandler struct { sto stores.Storage llm llm.Client toolreg *tools.Registry - toolExec *ToolExecutor + toolExec *agent.ToolExecutor rnr *runner.Runner } @@ -46,7 +47,7 @@ func InitChannels(r chi.Router, preset *aigc.Preset, sto stores.Storage, llmClie sto: sto, llm: llmClient, toolreg: toolreg, - toolExec: NewToolExecutor(toolreg), + toolExec: agent.NewToolExecutor(toolreg), rnr: runner.New(stores.NewSessionStore(sto), stores.NewHistoryStore(sto)), } @@ -177,76 +178,69 @@ func (chh *channelHandler) MessageHandler(p channel.Channel, msg *channel.Messag } // handleStreamingReply handles reply with streaming support (e.g., WeCom WebSocket). -// It uses a loop similar to chatStreamResponseLoop to handle tool calls. +// It delegates the agent loop to AgentLoop and manages the stream lifecycle on the handler side. func (chh *channelHandler) handleStreamingReply(ctx context.Context, p channel.Channel, msg *channel.Message, sr channel.StreamReplier, cs stores.Conversation) { // Build messages and get tools messages, tools := chh.buildChatMessagesAndTools(ctx, msg, cs) - // MaxLoopIterations limits tool call chain depth to prevent infinite loops (default: 5) - maxLoopIterations := settings.Current.MaxLoopIterations - iter := 0 - var fullAnswer string - var streamID string - - for { - iter++ - if iter > maxLoopIterations { - slog.Info("channel: streaming loop iteration limit reached", "maxIter", maxLoopIterations) - break - } + // Create AgentLoop to drive LLM calls + tool execution + loop := agent.NewAgentLoop(agent.AgentLoopConfig{ + LLM: chh.llm, + ToolExec: chh.toolExec, + MaxLoop: settings.Current.MaxLoopIterations, + }) - // First iteration: start stream immediately to notify platform we're processing - if streamID == "" { - var err error - streamID, err = sr.StartStream(ctx, msg.ReplyCtx, "正在思考...") - if err != nil { - slog.Error("channel: start stream failed", "err", err) - channelReplyError(p, msg, "AI processing failed") - return - } - } + var contentBuilder strings.Builder + var streamID string - // Do one streaming round (stream lifecycle managed by this function) - answer, toolCalls, err := chh.doChannelStream(ctx, p, msg, sr, streamID, messages, tools) + for event, err := range loop.Run(ctx, messages, tools) { if err != nil { - slog.Error("channel: stream round failed", "iter", iter, "err", err) - finishErr := sr.FinishStream(ctx, msg.ReplyCtx, streamID, translateLLMErrorToUser(err)) - if finishErr != nil { - slog.Warn("channel: finish stream after error failed, falling back to Reply", "err", finishErr) + slog.Error("channel: agent loop error", "err", err) + if streamID != "" { + if finishErr := sr.FinishStream(ctx, msg.ReplyCtx, streamID, translateLLMErrorToUser(err)); finishErr != nil { + slog.Warn("channel: finish stream after error failed, falling back to Reply", "err", finishErr) + channelReplyError(p, msg, "AI processing failed") + } + } else { channelReplyError(p, msg, "AI processing failed") } return } - slog.Info("channel: stream round done", - "iter", iter, - "answer_len", len(answer), - "toolCalls_len", len(toolCalls), - "streamID", streamID) - - // Only update fullAnswer if we got actual content - if answer != "" { - fullAnswer = answer - } - // No more tool calls, we're done - if len(toolCalls) == 0 { - break + // Skip tool result events — they are internal to the agent loop + if event.ToolResult != nil { + continue } - // Add assistant response to messages (with full answer content) - if fullAnswer != "" { - messages = append(messages, llm.Message{Role: llm.RoleAssistant, Content: fullAnswer}) + // Start stream on first non-empty delta + if streamID == "" { + if event.Delta == "" { + continue + } + var startErr error + streamID, startErr = sr.StartStream(ctx, msg.ReplyCtx, event.Delta) + if startErr != nil { + slog.Error("channel: start stream failed", "err", startErr) + channelReplyError(p, msg, "AI processing failed") + return + } + contentBuilder.WriteString(event.Delta) + slog.Info("channel: stream started", "streamID", streamID) + continue } - // Execute tool calls and update messages with results - evs, msgs := chh.toolExec.ExecuteToolCalls(ctx, messages, toolCalls, "") - messages = msgs - if len(evs) == 0 { - // 没有成功执行任何工具，跳出循环 - break + // Subsequent deltas: accumulate and send full content (WeCom overwrite semantics) + if event.Delta != "" { + contentBuilder.WriteString(event.Delta) + content := contentBuilder.String() + if err := sr.AppendStream(ctx, msg.ReplyCtx, streamID, content); err != nil { + slog.Warn("channel: append stream failed", "err", err) + } } } + fullAnswer := contentBuilder.String() + slog.Info("channel: streaming reply finishing", "streamID", streamID, "fullAnswer_len", len(fullAnswer)) @@ -271,58 +265,17 @@ func (chh *channelHandler) handleStreamingReply(ctx context.Context, p channel.C } } -// doChannelStream performs one streaming chat round, returns answer, tool calls, and streamID. -// The stream lifecycle (Start/Finish) is managed by the caller (handleStreamingReply). -func (chh *channelHandler) doChannelStream(ctx context.Context, p channel.Channel, msg *channel.Message, sr channel.StreamReplier, streamID string, messages []llm.Message, tools []llm.ToolDefinition) (string, []llm.ToolCall, error) { - var contentBuilder strings.Builder - var currentToolCalls []llm.ToolCall - chunkCount := 0 - - for result, err := range chh.llm.StreamChat(ctx, messages, tools) { - chunkCount++ - if err != nil { - slog.Warn("channel: stream error", "err", err) - break - } - - // Only send to channel when we have content; accumulate locally for WeCom overwrite semantics - if result.Delta != "" { - contentBuilder.WriteString(result.Delta) - content := contentBuilder.String() - if err := sr.AppendStream(ctx, msg.ReplyCtx, streamID, content); err != nil { - slog.Warn("channel: append stream failed", "err", err) - } - } - - // Capture tool calls when Done=true (LLM signaled end with tool calls) - if result.Done { - currentToolCalls = result.ToolCalls - slog.Debug("channel: stream Done received", "toolCalls_len", len(result.ToolCalls)) - } - } - - slog.Info("channel: doChannelStream result", - "chunkCount", chunkCount, - "content_len", contentBuilder.Len(), - "toolCalls_len", len(currentToolCalls), - "streamID", streamID) - - return contentBuilder.String(), currentToolCalls, nil -} - // handleRegularReply handles reply without streaming (non-WebSocket channels). func (chh *channelHandler) handleRegularReply(ctx context.Context, p channel.Channel, msg *channel.Message, cs stores.Conversation) { messages, tools := chh.buildChatMessagesAndTools(ctx, msg, cs) - // Execute the chat with tool call loop - exec := func(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, []llm.ToolCall, *llm.Usage, error) { - result, err := chh.llm.Chat(ctx, messages, tools) - if err != nil { - return "", nil, nil, err - } - return result.Content, result.ToolCalls, result.Usage, nil - } - answer, _, _, err := chh.executeToolCallLoop(ctx, messages, tools, exec) + // Non-streaming: use AgentLoop.RunNonStreaming + loop := agent.NewAgentLoop(agent.AgentLoopConfig{ + LLM: chh.llm, + ToolExec: chh.toolExec, + MaxLoop: settings.Current.MaxLoopIterations, + }) + answer, err := loop.RunNonStreaming(ctx, messages, tools) if err != nil { slog.Error("channel: chat execution failed", "channel", p.Name(), "error", err) @@ -349,11 +302,6 @@ func (chh *channelHandler) handleRegularReply(ctx context.Context, p channel.Cha } } -// executeToolCallLoop executes tool calls in a loop until no more tool calls -func (chh *channelHandler) executeToolCallLoop(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition, exec chatExecutor) (string, []llm.ToolCall, *llm.Usage, error) { - return chh.toolExec.ExecuteToolCallLoop(ctx, messages, tools, exec) -} - // channelReplyError sends an error message back to the channel. func channelReplyError(p channel.Channel, msg *channel.Message, errorText string) { ctx := context.Background() diff --git a/pkg/web/api/tool_executor.go b/pkg/web/api/tool_executor.go deleted file mode 100644 index df66bae..0000000 --- a/pkg/web/api/tool_executor.go +++ /dev/null @@ -1,112 +0,0 @@ -package api - -import ( - "context" - "encoding/json" - "time" - - "github.com/liut/morign/pkg/services/llm" - "github.com/liut/morign/pkg/services/tools" - toolsvc "github.com/liut/morign/pkg/services/tools" -) - -// chatExecutor 定义聊天执行函数类型，支持流式/非流式 -type chatExecutor func(ctx context.Context, messages []llm.Message, tools []llm.ToolDefinition) (string, []llm.ToolCall, *llm.Usage, error) - -// ToolExecutor 封装工具调用循环逻辑 -type ToolExecutor struct { - toolreg *tools.Registry -} - -// NewToolExecutor 创建 ToolExecutor -func NewToolExecutor(toolreg *tools.Registry) *ToolExecutor { - return &ToolExecutor{toolreg: toolreg} -} - -// ExecuteToolCallLoop 执行工具调用循环，直到无 tool calls -func (e *ToolExecutor) ExecuteToolCallLoop( - ctx context.Context, - messages []llm.Message, - tools []llm.ToolDefinition, - exec chatExecutor, -) (string, []llm.ToolCall, *llm.Usage, error) { - for { - answer, toolCalls, usage, err := exec(ctx, messages, tools) - if err != nil { - return "", nil, nil, err - } - - if len(toolCalls) == 0 { - return answer, nil, usage, nil - } - - evs, msgs := e.ExecuteToolCalls(ctx, messages, toolCalls, "") - messages = msgs - if len(evs) == 0 { - // 没有成功执行任何工具，跳出循环 - return answer, toolCalls, usage, nil - } - } -} - -// ExecuteToolCalls 执行单轮工具调用，返回事件列表和更新后的消息。 -func (e *ToolExecutor) ExecuteToolCalls(ctx context.Context, messages []llm.Message, toolCalls []llm.ToolCall, think string) ([]*llm.Event, []llm.Message) { - if len(toolCalls) == 0 { - return nil, messages - } - - messages = append(messages, llm.Message{ - Role: llm.RoleAssistant, - Thinking: think, - ToolCalls: toolCalls, - }) - - var events []*llm.Event - for _, tc := range toolCalls { - logger().Infow("chat", "toolCallID", tc.ID, "toolCallType", tc.Type, "toolCallName", tc.Function.Name) - - if tc.Type != "function" { - continue - } - - var parameters map[string]any - args := string(tc.Function.Arguments) - if args != "" && args != "{}" { - if err := json.Unmarshal(tc.Function.Arguments, ¶meters); err != nil { - logger().Infow("chat", "toolCallID", tc.ID, "args", args, "err", err) - continue - } - } - if parameters == nil { - parameters = make(map[string]any) - } - - content, err := e.toolreg.Invoke(ctx, tc.Function.Name, parameters) - if err != nil { - logger().Infow("invokeTool fail", "toolCallName", tc.Function.Name, "err", err) - continue - } - - logger().Infow("invokeTool ok", "toolCallName", tc.Function.Name, - "content", toolsvc.ResultLogs(content)) - toolResult := formatToolResult(content) - messages = append(messages, llm.Message{ - Role: llm.RoleTool, - Content: toolResult, - ToolCallID: tc.ID, - }) - - events = append(events, &llm.Event{ - ID: llm.NewEventID(), - Timestamp: time.Now(), - Author: tc.Function.Name, - ToolResult: &llm.ToolResult{ - CallID: tc.ID, - Name: tc.Function.Name, - Content: toolResult, - }, - }) - } - - return events, messages -}