diff --git a/backend/cmd/cyrene/main.go b/backend/cmd/cyrene/main.go index 82db849..4e62564 100644 --- a/backend/cmd/cyrene/main.go +++ b/backend/cmd/cyrene/main.go @@ -193,6 +193,7 @@ func main() { memExtractor := memory.NewExtractor(memSvc, &memory.LLMAdapterBridge{Adapter: memAdapter}) orch.SetMemoryExtractor(memExtractor) } + orch.SetToolExecutor(&toolExecAdapter{reg: reg}) logger.Infof("[orchestrator] pipeline ready") // === 11.5 Platform Bridge === @@ -330,6 +331,22 @@ func (a *perfAdapter) History() []map[string]any { return result } +// toolExecAdapter adapts the plugin registry to orchestrator.ToolExecutor. +type toolExecAdapter struct{ reg *plugin.Registry } + +func (t *toolExecAdapter) Tools() []llm.ToolSpec { + defs := t.reg.GetAllToolDefinitions() + specs := make([]llm.ToolSpec, len(defs)) + for i, d := range defs { + specs[i] = llm.ToolSpec{Name: d.Name, Description: d.Description, Parameters: d.Parameters} + } + return specs +} + +func (t *toolExecAdapter) Execute(ctx context.Context, name string, params map[string]any) (string, error) { + return t.reg.ExecuteTool(ctx, name, params) +} + // bridgeConfigAdapter adapts bridge.ConfigStore to api.BridgeConfigManager. type bridgeConfigAdapter struct{ *bridge.ConfigStore } func (a *bridgeConfigAdapter) GetAdapters() []map[string]any { diff --git a/backend/services/ai-core/llm/llm.go b/backend/services/ai-core/llm/llm.go index 0e8de79..c0ea633 100644 --- a/backend/services/ai-core/llm/llm.go +++ b/backend/services/ai-core/llm/llm.go @@ -84,7 +84,7 @@ func (a *Adapter) Chat(ctx context.Context, messages []model.LLMMessage) (*model start := time.Now() body := map[string]any{ "model": a.cfg.Model, - "messages": messages, + "messages": wireMessages(messages), } resp := &model.LLMResponse{Model: a.cfg.Model} @@ -106,14 +106,32 @@ func (a *Adapter) Chat(ctx context.Context, messages []model.LLMMessage) (*model return resp, err } +// ToolSpec describes a callable tool exposed to the LLM (function calling). +type ToolSpec struct { + Name string + Description string + // Parameters is a JSON Schema object describing the tool's input. + Parameters map[string]any +} + // ChatStream sends a chat completion request and returns a token channel. func (a *Adapter) ChatStream(ctx context.Context, messages []model.LLMMessage) (<-chan model.StreamToken, error) { + return a.ChatStreamWithTools(ctx, messages, nil) +} + +// ChatStreamWithTools is ChatStream with function-calling support. +// When the model requests tool calls, they are delivered on the final +// Done token (StreamToken.ToolCalls) instead of regular content tokens. +func (a *Adapter) ChatStreamWithTools(ctx context.Context, messages []model.LLMMessage, tools []ToolSpec) (<-chan model.StreamToken, error) { start := time.Now() body := map[string]any{ "model": a.cfg.Model, - "messages": messages, + "messages": wireMessages(messages), "stream": true, } + if len(tools) > 0 { + body["tools"] = wireTools(tools) + } ch := make(chan model.StreamToken, 64) go func() { @@ -203,6 +221,7 @@ func (a *Adapter) doStreamRequest(ctx context.Context, method, path string, body } scanner := bufio.NewScanner(resp.Body) + toolAcc := newToolCallAccumulator() for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" || !strings.HasPrefix(line, "data: ") { @@ -210,7 +229,7 @@ func (a *Adapter) doStreamRequest(ctx context.Context, method, path string, body } data := strings.TrimPrefix(line, "data: ") if data == "[DONE]" { - ch <- model.StreamToken{Done: true} + ch <- model.StreamToken{Done: true, ToolCalls: toolAcc.calls()} return nil } @@ -225,9 +244,10 @@ func (a *Adapter) doStreamRequest(ctx context.Context, method, path string, body if content, ok := delta["content"].(string); ok { ch <- model.StreamToken{Content: content} } + toolAcc.feed(delta) } if reason, ok := choice["finish_reason"].(string); ok && reason != "" { - ch <- model.StreamToken{FinishReason: reason, Done: true} + ch <- model.StreamToken{FinishReason: reason, Done: true, ToolCalls: toolAcc.calls()} return nil } } @@ -242,6 +262,122 @@ func min(a, b int) int { return b } +// wireMessages converts LLMMessages to the OpenAI wire format, +// expanding ToolCalls into the nested function-call structure. +func wireMessages(messages []model.LLMMessage) []map[string]any { + out := make([]map[string]any, len(messages)) + for i, m := range messages { + wm := map[string]any{"role": m.Role, "content": m.Content} + if m.Name != "" { + wm["name"] = m.Name + } + if m.ToolCallID != "" { + wm["tool_call_id"] = m.ToolCallID + } + if len(m.ToolCalls) > 0 { + tcs := make([]map[string]any, len(m.ToolCalls)) + for j, tc := range m.ToolCalls { + args, _ := json.Marshal(tc.Arguments) + tcs[j] = map[string]any{ + "id": tc.ID, + "type": "function", + "function": map[string]any{ + "name": tc.Name, + "arguments": string(args), + }, + } + } + wm["tool_calls"] = tcs + } + out[i] = wm + } + return out +} + +// wireTools converts ToolSpecs to the OpenAI tools array format. +func wireTools(tools []ToolSpec) []map[string]any { + out := make([]map[string]any, len(tools)) + for i, t := range tools { + params := t.Parameters + if params == nil { + params = map[string]any{"type": "object", "properties": map[string]any{}} + } + out[i] = map[string]any{ + "type": "function", + "function": map[string]any{ + "name": t.Name, + "description": t.Description, + "parameters": params, + }, + } + } + return out +} + +// toolCallAccumulator reassembles streamed tool_call deltas (indexed fragments +// with partial id/name/arguments) into complete ToolCalls. +type toolCallAccumulator struct { + order []int + byIdx map[int]*model.ToolCall + rawArg map[int]*strings.Builder +} + +func newToolCallAccumulator() *toolCallAccumulator { + return &toolCallAccumulator{byIdx: map[int]*model.ToolCall{}, rawArg: map[int]*strings.Builder{}} +} + +// feed consumes one streaming delta and merges any tool_calls fragments. +func (a *toolCallAccumulator) feed(delta map[string]any) { + tcs, ok := delta["tool_calls"].([]any) + if !ok { + return + } + for _, item := range tcs { + tcMap, ok := item.(map[string]any) + if !ok { + continue + } + idx := 0 + if f, ok := tcMap["index"].(float64); ok { + idx = int(f) + } + tc, exists := a.byIdx[idx] + if !exists { + tc = &model.ToolCall{} + a.byIdx[idx] = tc + a.rawArg[idx] = &strings.Builder{} + a.order = append(a.order, idx) + } + if id, ok := tcMap["id"].(string); ok && id != "" { + tc.ID = id + } + if fn, ok := tcMap["function"].(map[string]any); ok { + if name, ok := fn["name"].(string); ok && name != "" { + tc.Name = name + } + if args, ok := fn["arguments"].(string); ok { + a.rawArg[idx].WriteString(args) + } + } + } +} + +// calls returns the accumulated tool calls in index order, or nil if none. +func (a *toolCallAccumulator) calls() []model.ToolCall { + if len(a.order) == 0 { + return nil + } + calls := make([]model.ToolCall, 0, len(a.order)) + for _, idx := range a.order { + tc := a.byIdx[idx] + if raw := a.rawArg[idx].String(); raw != "" { + _ = json.Unmarshal([]byte(raw), &tc.Arguments) + } + calls = append(calls, *tc) + } + return calls +} + // ModelSelector routes requests to different models based on purpose. type ModelSelector struct { adapters map[model.LLMPurpose]*Adapter diff --git a/backend/services/ai-core/model/model.go b/backend/services/ai-core/model/model.go index 803847b..f45c4aa 100644 --- a/backend/services/ai-core/model/model.go +++ b/backend/services/ai-core/model/model.go @@ -42,6 +42,11 @@ type LLMMessage struct { Role string `json:"role"` // "system" | "user" | "assistant" | "tool" Content string `json:"content"` Name string `json:"name,omitempty"` + // ToolCallID links a tool result message to the call that produced it (role="tool"). + ToolCallID string `json:"tool_call_id,omitempty"` + // ToolCalls carries the calls requested by the assistant (role="assistant"). + // Marshaled to the OpenAI wire format by the llm package, hence "-". + ToolCalls []ToolCall `json:"-"` } // LLMResponse represents a non-streaming LLM response. @@ -54,9 +59,11 @@ type LLMResponse struct { // StreamToken is a single token from a streaming LLM response. type StreamToken struct { - Content string `json:"content"` - FinishReason string `json:"finish_reason,omitempty"` - Done bool `json:"done"` + Content string `json:"content"` + FinishReason string `json:"finish_reason,omitempty"` + Done bool `json:"done"` + // ToolCalls is set on the final Done token when the model requested tool calls. + ToolCalls []ToolCall `json:"tool_calls,omitempty"` } // ReviewSegment represents a parsed action/chat segment from LLM output. diff --git a/backend/services/ai-core/orchestrator/orchestrator.go b/backend/services/ai-core/orchestrator/orchestrator.go index eada172..7a09d21 100644 --- a/backend/services/ai-core/orchestrator/orchestrator.go +++ b/backend/services/ai-core/orchestrator/orchestrator.go @@ -20,6 +20,17 @@ type MemoryExtractor interface { ExtractFromConversation(ctx context.Context, userID string, conversation []model.Message) } +// ToolExecutor provides tool definitions and execution for function calling. +type ToolExecutor interface { + // Tools returns the tool specs exposed to the LLM. + Tools() []llm.ToolSpec + // Execute runs a tool by name with the given arguments. + Execute(ctx context.Context, name string, params map[string]any) (string, error) +} + +// maxToolRounds caps consecutive tool-call rounds to avoid runaway loops. +const maxToolRounds = 3 + // TraceHook is injected by the main app to record trace spans. type TraceHook func(name, label, status string, durationMs int64, detail string) @@ -35,6 +46,7 @@ type Pipeline struct { subMgr *subsession.Manager personaInjector *persona.Injector memoryExtractor MemoryExtractor + toolExecutor ToolExecutor } // NewPipeline creates a new dialogue pipeline. @@ -60,6 +72,11 @@ func (p *Pipeline) SetMemoryExtractor(e MemoryExtractor) { p.memoryExtractor = e } +// SetToolExecutor injects the tool execution capability (function calling). +func (p *Pipeline) SetToolExecutor(t ToolExecutor) { + p.toolExecutor = t +} + // ProcessInput processes a user message through the full pipeline. // Returns a channel of StreamToken for the LLM response. func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage string) (<-chan model.StreamToken, *model.Intent, error) { @@ -91,11 +108,17 @@ func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage stri llmMessages := p.ctxBuilder.Build(systemPrompt, sessionID, subResults, "") llmMessages = append(llmMessages, model.LLMMessage{Role: "user", Content: userMessage}) - // 6. LLM synthesis (streaming) + // 6. LLM synthesis (streaming, with tool calling when available) adapter := p.llmSelector.Select(model.PurposeChat) - tokenCh, err := adapter.ChatStream(ctx, llmMessages) - if err != nil { - return nil, intent, err + var tokenCh <-chan model.StreamToken + if p.toolExecutor != nil && len(p.toolExecutor.Tools()) > 0 { + tokenCh = p.streamWithTools(ctx, adapter, llmMessages) + } else { + var err error + tokenCh, err = adapter.ChatStream(ctx, llmMessages) + if err != nil { + return nil, intent, err + } } // 7. Store user message @@ -104,6 +127,77 @@ func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage stri return tokenCh, intent, nil } +// streamWithTools runs the LLM streaming loop with function calling. +// Content tokens are forwarded as they arrive; when the model requests tool +// calls, the tools are executed and their results are fed back for another +// round (up to maxToolRounds). The final round omits tools to force a text +// answer. +func (p *Pipeline) streamWithTools(ctx context.Context, adapter *llm.Adapter, messages []model.LLMMessage) <-chan model.StreamToken { + out := make(chan model.StreamToken, 64) + go func() { + defer close(out) + msgs := messages + tools := p.toolExecutor.Tools() + + for round := 0; round <= maxToolRounds; round++ { + useTools := tools + if round == maxToolRounds { + useTools = nil // Last chance: force a plain text answer + } + + tokenCh, err := adapter.ChatStreamWithTools(ctx, msgs, useTools) + if err != nil { + out <- model.StreamToken{Content: fmt.Sprintf("[Error: %v]", err), Done: true} + return + } + + var calls []model.ToolCall + for tok := range tokenCh { + if tok.Content != "" { + out <- model.StreamToken{Content: tok.Content} + } + if tok.Done { + calls = tok.ToolCalls + } + } + + if len(calls) == 0 { + out <- model.StreamToken{Done: true} + return + } + + // Record the assistant's tool-call request, then execute each call + // and append the results for the next round. + msgs = append(msgs, model.LLMMessage{Role: "assistant", ToolCalls: calls}) + for _, call := range calls { + result, err := p.toolExecutor.Execute(ctx, call.Name, call.Arguments) + if err != nil { + result = "Error: " + err.Error() + } + msgs = append(msgs, model.LLMMessage{ + Role: "tool", + ToolCallID: call.ID, + Name: call.Name, + Content: result, + }) + } + } + + out <- model.StreamToken{Done: true} + }() + return out +} + +// FinishRound records the assistant reply in the context store and triggers +// async memory extraction. Transport layers (WS hub, platform bridge) call it +// after the response stream completes. +func (p *Pipeline) FinishRound(sessionID, userID, assistantReply string) { + if assistantReply != "" { + p.ctxStore.Add(sessionID, model.Message{Role: "assistant", Content: assistantReply}) + } + p.PostProcess(sessionID, userID) +} + // PostProcess triggers async memory extraction after a conversation round completes. func (p *Pipeline) PostProcess(sessionID, userID string) { if p.memoryExtractor == nil { return } diff --git a/backend/services/ai-core/orchestrator/orchestrator_tools_test.go b/backend/services/ai-core/orchestrator/orchestrator_tools_test.go new file mode 100644 index 0000000..595537c --- /dev/null +++ b/backend/services/ai-core/orchestrator/orchestrator_tools_test.go @@ -0,0 +1,119 @@ +package orchestrator + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + ctxpkg "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/context" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/llm" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/persona" + "git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/subsession" +) + +// fakeToolExec records tool invocations and returns canned results. +type fakeToolExec struct { + executed []string + lastArgs map[string]any +} + +func (f *fakeToolExec) Tools() []llm.ToolSpec { + return []llm.ToolSpec{{ + Name: "iot_control", + Description: "控制智能家居设备", + Parameters: map[string]any{"type": "object", "properties": map[string]any{}}, + }} +} + +func (f *fakeToolExec) Execute(_ context.Context, name string, params map[string]any) (string, error) { + f.executed = append(f.executed, name) + f.lastArgs = params + return "设备已开启", nil +} + +// TestToolCallLoop verifies that tool_call deltas from the stream are +// reassembled, executed, and fed back so the model can produce a final answer. +func TestToolCallLoop(t *testing.T) { + var round atomic.Int32 + var secondReqHadToolMsg bool + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + n := round.Add(1) + + w.Header().Set("Content-Type", "text/event-stream") + if n == 1 { + // Round 1: request must carry tools; respond with fragmented tool_calls. + if !strings.Contains(string(body), `"tools"`) { + t.Errorf("round 1 request missing tools field") + } + io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"iot_control\",\"arguments\":\"{\\\"id\\\":\\\"light_1\\\",\"}}]}}]}\n\n") + io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"\\\"action\\\":\\\"on\\\"}\"}}]}}]}\n\n") + io.WriteString(w, "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n") + io.WriteString(w, "data: [DONE]\n\n") + return + } + + // Round 2: the tool result must be present in the conversation. + var req map[string]any + if err := json.Unmarshal(body, &req); err == nil { + for _, m := range req["messages"].([]any) { + msg := m.(map[string]any) + if msg["role"] == "tool" && msg["tool_call_id"] == "call_1" { + secondReqHadToolMsg = true + } + } + } + io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"灯已打开\"}}]}\n\n") + io.WriteString(w, "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n") + io.WriteString(w, "data: [DONE]\n\n") + })) + defer server.Close() + + llmCfg := llm.DefaultConfig() + llmCfg.BaseURL = server.URL + p := NewPipeline( + llm.NewModelSelector(llmCfg), + llm.NewSemanticCache(), + ctxpkg.NewStore(4000, 50), + subsession.NewManager(&subsession.GeneralProvider{}, &subsession.MemoryProvider{}), + persona.NewInjector(persona.BuiltinDefault()), + ) + exec := &fakeToolExec{} + p.SetToolExecutor(exec) + + tokenCh, _, err := p.ProcessInput(context.Background(), "s1", "打开客厅灯") + if err != nil { + t.Fatalf("ProcessInput: %v", err) + } + + var content strings.Builder + var sawDone bool + for tok := range tokenCh { + content.WriteString(tok.Content) + if tok.Done { + sawDone = true + } + } + + if !sawDone { + t.Error("stream never signaled Done") + } + if content.String() != "灯已打开" { + t.Errorf("final content = %q, want %q", content.String(), "灯已打开") + } + if len(exec.executed) != 1 || exec.executed[0] != "iot_control" { + t.Errorf("executed tools = %v, want [iot_control]", exec.executed) + } + if exec.lastArgs["id"] != "light_1" || exec.lastArgs["action"] != "on" { + t.Errorf("tool args = %v, want id=light_1 action=on", exec.lastArgs) + } + if !secondReqHadToolMsg { + t.Error("round 2 request missing tool result message") + } +}