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

package eino

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"time"

	"github.com/cloudwego/eino/components/tool"
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/flow/agent/react"
	"github.com/cloudwego/eino/schema"

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

// reactMaxStep 限制 ReAct 推理-工具循环的步数上限（控成本/时延）。
const reactMaxStep = 8

// streamHasToolCall 扫描模型整段流输出，任一片段含 tool call 即判定为工具调用。
// 比默认"只看首片段"鲁棒（兼容先吐文本/思考再给 tool call 的模型，如 deepseek）。
// 契约要求：返回前必须 Close 传入的流。
func streamHasToolCall(_ context.Context, sr *schema.StreamReader[*schema.Message]) (bool, error) {
	defer sr.Close()
	for {
		msg, err := sr.Recv()
		if err == io.EOF {
			return false, nil
		}
		if err != nil {
			return false, err
		}
		if len(msg.ToolCalls) > 0 {
			return true, nil
		}
	}
}

// mcpTool 把一个 MCP 工具（NATS 那头）适配成 Eino InvokableTool：
// 模型给的参数(JSON) + 运行时注入的绑定参数(uid/kb，不暴露给模型) 合并后经 NATS 调 MCP。
type mcpTool struct {
	info    *schema.ToolInfo
	mcpName string              // MCP 侧真实工具名（可与 info.Name 不同）
	subject func(string) string // contract.ToolSubjectGo/Py
	bind    map[string]any      // 运行时注入参数
	caller  ToolCaller
	taskID  string
	tr      *execTracer
}

func (m *mcpTool) Info(_ context.Context) (*schema.ToolInfo, error) { return m.info, nil }

// InvokableRun 执行一次工具调用：合并参数 → NATS 调 MCP → 返回观察给模型；
// 调用本身落一条 ExecEvent 轨迹（"模型自己调了哪个工具"在观测里可见）。
func (m *mcpTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.Option) (string, error) {
	args := map[string]any{}
	if argsJSON != "" {
		_ = json.Unmarshal([]byte(argsJSON), &args)
	}
	for k, v := range m.bind {
		args[k] = v
	}
	log.Printf("[react] 模型自主调用工具 %s (mcp=%s) task=%s args=%s", m.info.Name, m.mcpName, m.taskID, truncate(argsJSON, 120))
	end := m.tr.span("tool:"+m.mcpName, "tool", "模型自主调用 "+m.info.Name)
	cctx, cancel := context.WithTimeout(ctx, toolCallTimeout)
	defer cancel()
	res, err := m.caller.CallTool(cctx, m.subject(m.mcpName), &contract.ToolCall{Tool: m.mcpName, TaskID: m.taskID, Args: args})
	if err != nil {
		end("调用失败", err)
		return "工具调用失败：" + err.Error(), nil // 作为观察返回，不中断 ReAct
	}
	if res == nil || !res.OK {
		msg := "工具无结果"
		if res != nil && res.Error != "" {
			msg = res.Error
		}
		end(msg, nil)
		return msg, nil
	}
	end("入参 "+truncate(argsJSON, 120)+" → "+truncate(res.Content, 160), nil)
	return res.Content, nil
}

// agentTools 构建 ReAct 可用的工具集（按当前任务上下文绑定 uid/kb）。
// 工具名是给"模型看"的语义名；mcpName 是实际 NATS 调用名。context 参数（user_id）服务端注入，不暴露给模型。
func (o *Orchestrator) agentTools(b *board, taskID string, tr *execTracer) []tool.BaseTool {
	if o.tools == nil {
		return nil
	}
	kbBind := map[string]any{}
	if b.kb != "" {
		kbBind["kb"] = b.kb
	}
	return []tool.BaseTool{
		&mcpTool{
			mcpName: "wiki_search",
			subject: contract.ToolSubjectGo,
			caller:  o.tools, taskID: taskID, tr: tr,
			bind: kbBind,
			info: &schema.ToolInfo{
				Name: "wiki_search",
				Desc: "检索知识库，返回与查询最相关的资料片段。需要外部知识/事实依据时调用。",
				ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{
					"q": {Type: schema.String, Desc: "检索查询语句", Required: true},
				}),
			},
		},
		&mcpTool{
			mcpName: "memory_get",
			subject: contract.ToolSubjectGo,
			caller:  o.tools, taskID: taskID, tr: tr,
			bind: map[string]any{"user_id": b.uid},
			info: &schema.ToolInfo{
				Name: "recall_user_memory",
				Desc: "召回当前用户的长期画像与偏好（称呼/职业/回答偏好等）。需要个性化、了解“我是谁”时调用。",
				ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{}),
			},
		},
	}
}

// runReactAgent 执行带"自主工具"的 agent 节点：模型在 ReAct 循环里自行决定调哪些 MCP 工具。
// 模型不支持函数调用 / 无工具时降级回普通 runAgent。最终答复流式回流；工具调用由适配器落轨迹。
func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *board, system string, n dsl.Node, tr *execTracer, node string) {
	tcm := o.pool.ToolCallingModel()
	tools := o.agentTools(b, taskID, tr)
	if tcm == nil || len(tools) == 0 {
		tr.info(node, "system", "ReAct 降级", "模型不支持函数调用或无可用工具，退回普通对话")
		o.runAgent(ctx, taskID, b, system, tr, node)
		return
	}

	ag, err := react.NewAgent(ctx, &react.AgentConfig{
		ToolCallingModel: tcm,
		ToolsConfig:      compose.ToolsNodeConfig{Tools: tools},
		MaxStep:          reactMaxStep,
		// 默认检查器只看首个流片段；deepseek 等常先吐文本再给 tool call → 漏判。
		// 改成扫描整段流，任一片段含 tool call 即判定为工具调用。
		StreamToolCallChecker: streamHasToolCall,
	})
	if err != nil {
		tr.emit(node, "model", "error", "构建 ReAct 智能体", err.Error(), 0)
		o.runAgent(ctx, taskID, b, system, tr, node)
		return
	}

	rc := &RunCtx{
		UserID: b.uid, SessionID: b.sid,
		System: firstNonEmpty(system, defaultAgentSystem),
		Query:  b.query,
		// 自主 agent 不预注入画像：让它经 recall_user_memory 工具按需自取（否则模型直接答、不调工具）。
		Profile: "",
		History: b.history,
		ToolOut: append(append([]string{}, b.toolOut...), b.refs...),
	}
	msgs, _ := buildMessages(ctx, rc)

	tr.emit(node, "model", "start", "ReAct 智能体（自主调工具）", fmt.Sprintf("%d 个工具可用", len(tools)), 0)
	t0 := time.Now()
	sr, err := ag.Stream(ctx, msgs)
	if err != nil {
		tr.emit(node, "model", "error", "ReAct 智能体", err.Error(), time.Since(t0).Milliseconds())
		return
	}
	defer sr.Close()

	chunks := 0
	for {
		chunk, rerr := sr.Recv()
		if rerr == io.EOF {
			break
		}
		if rerr != nil {
			tr.emit(node, "model", "error", "ReAct 智能体", rerr.Error(), time.Since(t0).Milliseconds())
			return
		}
		if chunk.Content == "" {
			continue // 工具调用片段无正文，跳过；正文只来自模型答复
		}
		safe, _ := harness.RedactSecrets(chunk.Content)
		_ = o.sink.PublishToken(taskID, []byte(safe))
		b.answer += safe
		chunks++
	}
	tr.emit(node, "model", "end", "ReAct 智能体",
		fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(b.answer))), time.Since(t0).Milliseconds())
}