19904408334
cef837a35c
- 新增同会话+同工作流最近执行失败且参数一致时断点续跑逻辑 - exec_workflow/exec_chat 新增 error 字段存储原始错误，error_message 仅存友好提示 - 新增 UpdateExecChatReq 与 exec_chat_dao Update 方法 - 新增 GetLatestBySessionAndFlow 查询最近执行记录 - 修正 ListDates 分组与排序 SQL 表达式 - 新增 pipeline 配置结构，删除旧设计文档
260 lines
9.7 KiB
Go
260 lines
9.7 KiB
Go
// Package runner 工具框架的"使用面"：ReAct 执行循环。
|
||
// 模型调用按模型响应类型分流：流式走 gateway.ModelCallStream（逐 chunk 推增量），同步/异步走 gateway.ModelCallResult；
|
||
// 本包只保留"模型思考→调用工具→观察结果"的循环编排，工具的"使用"统一走 tools.Default.Call。
|
||
package runner
|
||
|
||
import (
|
||
"ai-agent/gateway"
|
||
"ai-agent/workflow/consts/model"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/tools"
|
||
)
|
||
|
||
// ReActEventType ReAct 过程事件类型（WebSocket 场景逐条推送，HTTP 同步场景不设置 OnEvent 则忽略）
|
||
type ReActEventType string
|
||
|
||
const (
|
||
ReActEventRoundStart ReActEventType = "round_start"
|
||
ReActEventModelCall ReActEventType = "model_call" // 模型思考（step 开始）
|
||
ReActEventToolCall ReActEventType = "tool_call" // 模型请求调用工具
|
||
ReActEventToolResult ReActEventType = "tool_result" // 工具返回结果
|
||
ReActEventAnswer ReActEventType = "answer" // 最终回答
|
||
ReActEventAnswerChunk ReActEventType = "answer_chunk" // 回答内容增量（逐 chunk）
|
||
ReActEventReasoningChunk ReActEventType = "reasoning_chunk" // 思考内容增量（逐 chunk）
|
||
ReActEventError ReActEventType = "error"
|
||
)
|
||
|
||
// ReActEvent ReAct 过程事件，按发生顺序回调
|
||
type ReActEvent struct {
|
||
Type ReActEventType
|
||
Id int64
|
||
Step int
|
||
MaxStep int
|
||
Description string // 工具用途说明（tool_call/tool_result 推给前端展示，不暴露工具名/参数）
|
||
Answer string
|
||
Delta string // 流式文本增量（ReActEventAnswerChunk / ReActEventReasoningChunk 用）
|
||
Message string
|
||
Error string
|
||
}
|
||
|
||
// ReActAgent 实现 ReAct 模式的智能体：模型思考→调用工具→观察工具结果→重复→最终回答。
|
||
// 工具的"使用"统一走 tools.Default.Call：未知工具由 Server 返回结构化 not_found，
|
||
// 无需在 runner 内维护工具查找表。
|
||
type ReActAgent struct {
|
||
ModelId int64
|
||
SessionId string
|
||
ToolList []*tools.Tool
|
||
SystemPrompt string
|
||
MaxStep int
|
||
// OnEvent 过程回调（WebSocket 流式推送用；HTTP 同步调用不设置，忽略事件）
|
||
OnEvent func(ReActEvent)
|
||
// TotalTokens 本次 ReAct 循环累计 token 消耗（供调用方落库）
|
||
TotalTokens int64
|
||
// TotalCost 本次 ReAct 循环累计费用（元，各步模型调用返回的 cost 求和，供调用方落库）
|
||
TotalCost float64
|
||
}
|
||
|
||
// NewReActAgent 创建 ReAct 智能体
|
||
func NewReActAgent(modelId int64, sessionId string, toolList []*tools.Tool, systemPrompt string, maxStep int) *ReActAgent {
|
||
return &ReActAgent{
|
||
ModelId: modelId,
|
||
SessionId: sessionId,
|
||
ToolList: toolList,
|
||
SystemPrompt: systemPrompt,
|
||
MaxStep: maxStep,
|
||
}
|
||
}
|
||
|
||
// Run 执行 ReAct 循环
|
||
// 标准流程: 思考 → 行动(调用工具) → 观察(工具结果) → 重复 → 最终回答
|
||
func (a *ReActAgent) Run(ctx context.Context, userInput string) (string, error) {
|
||
// 模型须为已启用且勾选「聊天模型」的对话模型，再按其响应类型分流调用：
|
||
// 流式走 ModelCallStream（逐 chunk 推增量），同步/异步走 ModelCallResult（一次返回）
|
||
modelInfo, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: a.ModelId})
|
||
if err != nil {
|
||
return "", fmt.Errorf("获取模型配置失败: %w", err)
|
||
}
|
||
if modelInfo.ModelManage.Enabled == nil || !*modelInfo.ModelManage.Enabled {
|
||
return "", fmt.Errorf("模型 [%s] 未启用", modelInfo.ModelManage.ModelName)
|
||
}
|
||
if modelInfo.ModelManage.ChatModel == nil || !*modelInfo.ModelManage.ChatModel {
|
||
return "", fmt.Errorf("模型 [%s] 不是对话模型，请在模型配置中勾选「聊天模型」", modelInfo.ModelManage.ModelName)
|
||
}
|
||
isStream := *modelInfo.ModelManage.ResponseType == *model.ResponseTypeStream.Code()
|
||
|
||
messages := map[string]any{
|
||
"user_prompt": userInput,
|
||
"system_prompt": a.SystemPrompt,
|
||
"tools": a.rawTools(a.ToolList),
|
||
}
|
||
for step := 0; step < a.MaxStep; step++ {
|
||
a.emit(ReActEvent{Type: ReActEventModelCall, Step: step + 1, MaxStep: a.MaxStep})
|
||
var result *gateway.ModelCallRes
|
||
if isStream {
|
||
result, err = gateway.ModelCallStream(ctx, a.ModelId, a.SessionId, nil, messages, func(chunk map[string]any) error {
|
||
// 文本增量从 content 子对象取，字段名由网关端结构体统一管理
|
||
if content, ok := chunk["content"].(map[string]any); ok {
|
||
for _, v := range content {
|
||
if s := gconv.String(v); s != "" {
|
||
a.emit(ReActEvent{Type: ReActEventAnswerChunk, Delta: s})
|
||
}
|
||
}
|
||
}
|
||
// 思考内容增量从 reasoningContent 字段取（model-gateway 按 ResponseBusinessFieldMapping 配置返回）
|
||
if s := gconv.String(chunk["reasoningContent"]); s != "" {
|
||
a.emit(ReActEvent{Type: ReActEventReasoningChunk, Delta: s})
|
||
}
|
||
return nil
|
||
})
|
||
} else {
|
||
// 同步/异步对话模型：结果一次返回，无增量事件（异步 msgTopic 为空时由 gateway.ModelCallResult 自动生成）
|
||
result, err = gateway.ModelCallResult(ctx, a.ModelId, modelInfo.ModelManage.ResponseType, a.SessionId, nil, messages)
|
||
}
|
||
// 出错时 result 仍可能携带已累计的 token/cost（流中断返回 res），先累加再判错
|
||
if result != nil {
|
||
a.TotalTokens += result.TotalTokens
|
||
a.TotalCost += result.Cost
|
||
}
|
||
if err != nil {
|
||
// 前端终止/上下文取消：静默停止，不推错误事件
|
||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||
return "", err
|
||
}
|
||
a.emit(ReActEvent{Type: ReActEventError, Message: "模型调用失败", Error: err.Error()})
|
||
return "", fmt.Errorf("step %d: model call failed: %w", step, err)
|
||
}
|
||
var content string
|
||
var toolCalls []gateway.ModelTool
|
||
if isStream {
|
||
for _, v := range gconv.Map(result.Content) {
|
||
content += v.(string)
|
||
}
|
||
toolCalls = result.Tools
|
||
} else {
|
||
// 同步响应 Content 按 responseBodyMapping 抽取（content/toolCalls/finishReason），工具在 toolCalls 里
|
||
content, toolCalls = syncChatResult(result.Content)
|
||
}
|
||
// 无工具调用 → 最终回答
|
||
if len(toolCalls) == 0 {
|
||
return content, nil
|
||
}
|
||
messages["assistant_prompt"] = content
|
||
for _, tool := range toolCalls {
|
||
fn := gconv.Map(tool.Function)
|
||
name := gconv.String(fn["name"])
|
||
arguments := gconv.String(fn["arguments"])
|
||
desc := a.toolDescription(name)
|
||
a.emit(ReActEvent{Type: ReActEventToolCall, Description: desc})
|
||
// 执行每个工具调用
|
||
toolMsg := a.executeToolCall(ctx, name, arguments)
|
||
a.emit(ReActEvent{Type: ReActEventToolResult, Description: desc})
|
||
messages["tool_prompt"] = toolMsg
|
||
messages["tool_id"] = tool.Id
|
||
}
|
||
}
|
||
err = fmt.Errorf("max steps reached %d, generation incomplete", a.MaxStep)
|
||
a.emit(ReActEvent{Type: ReActEventError, Message: "最大步数达到，生成失败", Error: err.Error()})
|
||
return "", err
|
||
}
|
||
|
||
// syncChatResult 解析同步对话响应 Content（model-gateway 按 responseBodyMapping 抽取，固定
|
||
// content/toolCalls/finishReason 三个 key）：返回最终文本与工具调用列表（同步响应工具在 toolCalls 里，res.Tools 为空）。
|
||
func syncChatResult(content map[string]any) (string, []gateway.ModelTool) {
|
||
text := gconv.String(content["content"])
|
||
raw := content["toolCalls"]
|
||
if raw == nil {
|
||
raw = content["tools"]
|
||
}
|
||
var toolCalls []gateway.ModelTool
|
||
if arr, ok := raw.([]any); ok {
|
||
for _, t := range arr {
|
||
m := gconv.Map(t)
|
||
if m == nil {
|
||
continue
|
||
}
|
||
var tool gateway.ModelTool
|
||
tool.Id = gconv.String(m["id"])
|
||
tool.Type = gconv.String(m["type"])
|
||
if fn := gconv.Map(m["function"]); fn != nil {
|
||
tool.Function.Name = gconv.String(fn["name"])
|
||
tool.Function.Arguments = gconv.String(fn["arguments"])
|
||
}
|
||
toolCalls = append(toolCalls, tool)
|
||
}
|
||
}
|
||
return text, toolCalls
|
||
}
|
||
|
||
// emit 触发过程事件回调；未设置 OnEvent 时忽略（HTTP 同步调用路径）
|
||
func (a *ReActAgent) emit(ev ReActEvent) {
|
||
if a.OnEvent != nil {
|
||
a.OnEvent(ev)
|
||
}
|
||
}
|
||
|
||
// executeToolCall 执行单个工具调用，返回回灌给模型的工具结果消息。
|
||
// 未知工具由 tools.Default.Call 返回结构化 not_found；工具业务失败进 ToolResult.Code。
|
||
func (a *ReActAgent) executeToolCall(ctx context.Context, name, arguments string) string {
|
||
if arguments == "" {
|
||
return "工具参数为空"
|
||
}
|
||
|
||
var args map[string]any
|
||
if err := json.Unmarshal([]byte(arguments), &args); err != nil {
|
||
return fmt.Sprintf("参数解析失败: %v", err)
|
||
}
|
||
|
||
res, callErr := tools.Default.Call(ctx, name, args)
|
||
|
||
switch {
|
||
case callErr != nil:
|
||
msg := fmt.Sprintf("工具执行失败: %v", callErr)
|
||
return msg
|
||
case res.Code != 0:
|
||
return res.Message
|
||
default:
|
||
dataJSON, err := json.Marshal(res.Data)
|
||
if err != nil {
|
||
return fmt.Sprintf("工具结果序列化失败: %v", err)
|
||
}
|
||
return string(dataJSON)
|
||
}
|
||
}
|
||
|
||
// toolDescription 按工具名从 ToolList 中查找用途说明（推给前端展示用，不暴露工具名/参数）
|
||
func (a *ReActAgent) toolDescription(name string) string {
|
||
for _, t := range a.ToolList {
|
||
if t.Name == name {
|
||
return t.Description
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// RawTools 将工具定义转为 OpenAI 原始数组，parameters 保持 JSON Schema 原样。
|
||
// businessParams 里传 tools 必须用它转换——直接传 []*tools.Tool 会被 json.Marshal 序列化 Func 字段而失败。
|
||
func (a *ReActAgent) rawTools(toolList []*tools.Tool) []any {
|
||
raw := make([]any, 0, len(toolList))
|
||
for _, t := range toolList {
|
||
params := t.Parameters
|
||
if params == nil {
|
||
params = map[string]any{}
|
||
}
|
||
raw = append(raw, map[string]any{
|
||
"type": "function",
|
||
"function": map[string]any{
|
||
"name": t.Name,
|
||
"description": t.Description,
|
||
"parameters": params,
|
||
},
|
||
})
|
||
}
|
||
return raw
|
||
}
|