sundynix-agentix/sundynix-dispatcher/internal/eino/orchestrator.go

// Package eino 封装基于 CloudWeGo Eino 的 Agent 图编排引擎。
package eino

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"log"
	"strings"
	"time"

	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/schema"

	"github.com/sundynix/sundynix-dispatcher/internal/dsl"
	"github.com/sundynix/sundynix-dispatcher/internal/harness"
	"github.com/sundynix/sundynix-dispatcher/internal/llm"
	"github.com/sundynix/sundynix-shared/contract"
)

// TokenSink 是 Token 流回流出口（由 NATS bus 实现）。
type TokenSink interface {
	PublishToken(taskID string, token []byte) error
	CompleteStream(taskID string) error
}

// ToolCaller 经 NATS 调起第 5 层 MCP 工具（由 NATS bus 实现）。
type ToolCaller interface {
	CallTool(ctx context.Context, subject string, call *contract.ToolCall) (*contract.ToolResult, error)
}

// 工具调用超时；超时即降级（不带工具上下文继续推理）。
const toolCallTimeout = 3 * time.Second

// Orchestrator 把 DSL 任务交给编译好的 Eino 图执行（记忆召回 → 注入 → 流式）。
type Orchestrator struct {
	breaker *harness.CircuitBreaker
	sink    TokenSink
	tools   ToolCaller
	run     compose.Runnable[*contract.Task, *schema.Message]
}

// NewOrchestrator 构建并编译记忆增强图。
func NewOrchestrator(pool *llm.Pool, breaker *harness.CircuitBreaker, sink TokenSink, tools ToolCaller) (*Orchestrator, error) {
	o := &Orchestrator{breaker: breaker, sink: sink, tools: tools}
	run, err := buildGraph(context.Background(), pool, o.fetchMemory, o.fetchHistory)
	if err != nil {
		return nil, err
	}
	o.run = run
	return o, nil
}

// Handle 消费一个任务：执行 Eino 图，把 Token 流回流到 sundynix.streams.<id>。
func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error {
	if !o.breaker.Allow() {
		log.Printf("[eino] circuit open, drop task %s", t.ID)
		return nil
	}
	log.Printf("[eino] task %s received (graph=%d bytes), running graph...", t.ID, len(t.Graph))

	stream, err := o.run.Stream(ctx, t)
	if err != nil {
		log.Printf("[eino] task %s graph error: %v", t.ID, err)
		_ = o.sink.CompleteStream(t.ID)
		o.breaker.Report(false)
		return err
	}
	defer stream.Close()

	n := 0
	var answer strings.Builder
	for {
		chunk, rerr := stream.Recv()
		if errors.Is(rerr, io.EOF) {
			break
		}
		if rerr != nil {
			log.Printf("[eino] task %s stream recv error: %v", t.ID, rerr)
			break
		}
		if chunk == nil || chunk.Content == "" {
			continue
		}
		if perr := o.sink.PublishToken(t.ID, []byte(chunk.Content)); perr != nil {
			log.Printf("[eino] publish token failed: %v", perr)
			break
		}
		answer.WriteString(chunk.Content)
		n++
	}

	if cerr := o.sink.CompleteStream(t.ID); cerr != nil {
		log.Printf("[eino] complete stream failed: %v", cerr)
	}
	log.Printf("[eino] task %s done, %d tokens streamed", t.ID, n)
	o.breaker.Report(true)

	// 写回阶段：流已排空（= 模型生成结束），此处离开热路径、异步落历史 + 抽取记忆。
	// 注：流式节点用 OnEndWithStreamOutput 而非 OnEndFn，故不走回调而在此触发。
	go o.memorize(t, answer.String())
	return nil
}

// fetchMemory 经 MCP memory_get 工具召回用户常驻画像。
// 工具不可用/超时/无 user_id 时返回空串，降级为无记忆推理（不阻断主流程）。
func (o *Orchestrator) fetchMemory(ctx context.Context, userID, _ string) string {
	if o.tools == nil || userID == "" {
		return ""
	}
	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
	defer cancel()
	res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("memory_get"), &contract.ToolCall{
		Tool: "memory_get",
		Args: map[string]any{"user_id": userID},
	})
	if err != nil {
		log.Printf("[eino] memory_get unavailable for %s, degrade: %v", userID, err)
		return ""
	}
	if !res.OK {
		log.Printf("[eino] memory_get error for %s: %s", userID, res.Error)
		return ""
	}
	log.Printf("[eino] memory_get ok for %s: %s", userID, res.Content)
	return res.Content
}

// fetchHistory 经 MCP history_get 工具召回会话短期多轮历史，转为 Eino 消息。
// 工具不可用/无 session 时返回空，降级为无历史（不阻断主流程）。
func (o *Orchestrator) fetchHistory(ctx context.Context, sessionID string) []*schema.Message {
	if o.tools == nil || sessionID == "" {
		return nil
	}
	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
	defer cancel()
	res, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("history_get"), &contract.ToolCall{
		Tool: "history_get",
		Args: map[string]any{"session_id": sessionID},
	})
	if err != nil || res == nil || !res.OK || res.Content == "" {
		return nil
	}
	var turns []struct {
		Role    string `json:"role"`
		Content string `json:"content"`
	}
	if json.Unmarshal([]byte(res.Content), &turns) != nil {
		return nil
	}
	msgs := make([]*schema.Message, 0, len(turns))
	for _, tn := range turns {
		if tn.Role == "assistant" {
			msgs = append(msgs, schema.AssistantMessage(tn.Content, nil))
		} else {
			msgs = append(msgs, schema.UserMessage(tn.Content))
		}
	}
	if len(msgs) > 0 {
		log.Printf("[eino] history_get ok for %s: %d 条历史", sessionID, len(msgs))
	}
	return msgs
}

// memorize 写回阶段：把本轮对话落进短期历史，并（TODO）抽取长期偏好记忆。
// 异步执行，离开热路径。
func (o *Orchestrator) memorize(t *contract.Task, answer string) {
	uid, _ := t.Meta[contract.MetaUserID].(string)
	sid, _ := t.Meta[contract.MetaSessionID].(string)
	if sid != "" && o.tools != nil {
		o.appendHistory(sid, "user", dsl.Compile(t.Graph).Query) // 落真实用户输入，而非 DSL 原文
		o.appendHistory(sid, "assistant", answer)
		log.Printf("[eino] (writeback) task %s 已落会话历史 session=%s", t.ID, sid)
	}
	if uid != "" {
		log.Printf("[eino] (writeback) task %s 待抽取 user=%s 的新偏好记忆", t.ID, uid)
		// TODO: 抽取 LLM → 去重/更新 → memory_upsert
	}
}

func (o *Orchestrator) appendHistory(sessionID, role, content string) {
	cctx, cancel := context.WithTimeout(context.Background(), toolCallTimeout)
	defer cancel()
	if _, err := o.tools.CallTool(cctx, contract.ToolSubjectGo("history_append"), &contract.ToolCall{
		Tool: "history_append",
		Args: map[string]any{"session_id": sessionID, "role": role, "content": content},
	}); err != nil {
		log.Printf("[eino] history_append failed: %v", err)
	}
}