feat(ai-core): wire function calling into the LLM pipeline

- llm: ChatStreamWithTools sends OpenAI-format tools; streamed tool_call
  deltas are reassembled by index into complete ToolCalls on the Done token
- model: LLMMessage gains ToolCallID/ToolCalls; StreamToken gains ToolCalls
- orchestrator: new ToolExecutor interface and streamWithTools loop —
  execute requested tools, feed results back, up to 3 rounds, final round
  forces a plain text answer; normal chats still stream token by token.
  Also adds Pipeline.FinishRound (assistant reply recording + post-round
  hook), wired to transports in the next commit.
- main: inject the plugin registry via toolExecAdapter

Includes an httptest-based test covering delta reassembly, tool
execution, and result feedback.

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 {

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

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.

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 }

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")
+	}
+}