From cef837a35c0c5afedc45c071c7d7abd9a6b29500 Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Fri, 21 Aug 2026 09:54:06 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E6=96=AD=E7=82=B9=E7=BB=AD=E8=B7=91=E5=B9=B6=E6=8B=86?= =?UTF-8?q?=E5=88=86=E9=94=99=E8=AF=AF=E4=BF=A1=E6=81=AF=E5=AD=98=E5=82=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增同会话+同工作流最近执行失败且参数一致时断点续跑逻辑 - exec_workflow/exec_chat 新增 error 字段存储原始错误，error_message 仅存友好提示 - 新增 UpdateExecChatReq 与 exec_chat_dao Update 方法 - 新增 GetLatestBySessionAndFlow 查询最近执行记录 - 修正 ListDates 分组与排序 SQL 表达式 - 新增 pipeline 配置结构，删除旧设计文档 --- config.yml | 11 +- ...26-08-11-shots-timeline-pipeline-design.md | 226 --------- go.mod | 15 +- go.sum | 4 + main.go | 2 + tools/runner/react_agent.go | 3 +- update.sql | 16 +- workflow/consts/node/node_template.go | 88 +++- .../controller/session/session_controller.go | 5 + workflow/dao/session/exec_chat_dao.go | 8 + workflow/dao/session/exec_workflow_dao.go | 34 +- workflow/model/dto/flow/flow_execution_dto.go | 2 + workflow/model/dto/session/exec_chat_dto.go | 15 +- .../model/dto/session/exec_workflow_dto.go | 6 +- .../dto/session/exec_workflow_result_dto.go | 4 - workflow/model/dto/session/session_dto.go | 11 +- workflow/model/entity/exec_chat.go | 5 +- workflow/model/entity/exec_workflow.go | 5 +- workflow/model/entity/flow_user.go | 43 +- .../service/flow/flow_checkpoint_store.go | 2 - workflow/service/flow/flow_ws_exec.go | 54 ++- workflow/service/flow/lambda_node.go | 137 ++++-- workflow/service/flow/lambda_node_util.go | 185 +++++++- .../service/flow/lambda_script_transcribe.go | 30 +- workflow/service/flow/lambda_value_source.go | 449 +++++++++++++++++- .../flow/processor/builtin/media/media.go | 72 ++- .../builtin/split_batch/split_batch.go | 186 ++++---- .../builtin/split_segment/split_segment.go | 161 +++++++ .../split_shots_pipeline/pipeline/config.go | 100 ++++ .../split_shots_pipeline.go | 1 + workflow/service/flow/processor/processor.go | 1 + workflow/service/flow/react_ws_exec.go | 89 +++- workflow/service/node/node_library_service.go | 57 +++ workflow/service/session/session_service.go | 7 + 34 files changed, 1519 insertions(+), 515 deletions(-) delete mode 100644 docs/superpowers/specs/2026-08-11-shots-timeline-pipeline-design.md create mode 100644 workflow/service/flow/processor/builtin/split_segment/split_segment.go create mode 100644 workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/config.go diff --git a/config.yml b/config.yml index 0d0ea0d..cf1339e 100644 --- a/config.yml +++ b/config.yml @@ -64,5 +64,12 @@ consul: jaeger: addr: 192.168.0.83:4318 -# 文件上传服务地址，与oss模块minio中的endpoint一致 -filePrefix: "http://192.168.0.83:9000" +nats: + addr: 192.168.0.83 + port: 4222 + +# 文件上传服务地址，cdn访问地址 +filePrefix: "http://cdn.redpowerfuture.com" + +# 文件上传服务地址，minio内网访问地址 +minioPrefix: "http://192.168.0.83:9000" diff --git a/docs/superpowers/specs/2026-08-11-shots-timeline-pipeline-design.md b/docs/superpowers/specs/2026-08-11-shots-timeline-pipeline-design.md deleted file mode 100644 index 2d1a1a5..0000000 --- a/docs/superpowers/specs/2026-08-11-shots-timeline-pipeline-design.md +++ /dev/null @@ -1,226 +0,0 @@ -# 分镜 → 时间线 → 模型语言 Pipeline 设计 - -日期：2026-08-11 -状态：已评审（用户确认） -范围：ai-agent 新增独立纯函数包 `video/pipeline/` - -## 1. 背景与动机 - -`script_transcribe` 节点产出 `[{"shots":[...]}]` 后，视频生成需要把分镜镜头转成视频模型能理解的输入。video-factory 项目已有完整链路：**时间线分段 → prompt（实体名→characterN 替换）→ 按 schema 嵌套的完整视频 API 请求体**。 - -ai-agent 现有 `video/plan/` 已覆盖"重建时间线 + 拆段 + 中文 prompt + 扁平 modelCall 参数"，但与 video-factory 存在差异，且缺少三件事： - -1. **时间线口径**：现有 `NormalizeShots` 丢弃 AI 时间戳、按台词字数重建时长（保留此口径，用户选定"重建但回写对齐"）； -2. **跨边界镜头不切分**：跨段边界的镜头原样保留、时间码不回写对齐（本方案改为按段边界切分 + 短残片并入）； -3. **模型语言转换缺失**： - - prompt 内实体名未替换成 `characterN` 等 token（`reference_labels` 单独传，靠模型自行理解）； - - 无按 schema 嵌套的完整视频 API 请求体（`video-factory` 的 `BuildSchemaRequest` 等价物）。 - -用户决策：**新流程从零设计**，做成**独立纯函数包**（不依赖 `plan`、不碰 workflow、零 I/O），**外部直接传 shots**，目标视频 API **先做通用抽象**（schema 可配），模型语言覆盖 **prompt 文本层 + 完整请求体层**。 - -## 2. 目标与非目标 - -### 目标 -- 提供一个纯函数入口：`shots → 时间线（重建+回写对齐）→ 分段 → prompt（token 替换）→ 完整视频 API 请求体` -- 通用抽象：模型 schema 可配置，未配置时用默认火山风格结构 -- 可独立单测，零 I/O、零 workflow 依赖 - -### 非目标（本阶段不做） -- 不接入 workflow 节点 / 前后置处理器（后续可写薄适配层） -- 不调用模型网关 / 媒体服务（`video/plan` 与 `video/` 根包负责） -- 不做视频合并（concat/merge）、不做首帧联动（serial） -- 不重构现有 `video/plan/` 或 `script_transcribe` - -## 3. 包结构与依赖 - -- 新包：`ai-agent/video/pipeline/` -- 依赖：仅 `ai-agent/video/domain`（复用 `domain.Shot` 作为输入契约）+ Go 标准库（`encoding/json`、`strings`、`fmt` 等） -- **不 import** `plan`、workflow、gateway、media 等 -- 内部自带：时间线重建、超长镜头切分、段时长计算、回写对齐、prompt 构建、schema 请求体构建 - -文件规划（建议）： -``` -video/pipeline/ - pipeline.go // Input/Segment/Output 契约 + Run 编排入口 - timeline.go // ① 时间线构建：rebuildDurations / splitOversized / calcSegmentDurations / alignToSegments - prompt.go // ② prompt 构建：实体名→token 替换、截断 - request.go // ③ 请求体构建：schema 嵌套、默认 schema、media 组装 - *_test.go // 各阶段单测 + 用户样本 golden test -``` - -## 4. 数据契约 - -```go -// Input 外部直接传入的生成请求参数。 -type Input struct { - Shots []domain.Shot // 镜头脚本（中文键或英文键均可，调用方已归一为 domain.Shot） - TotalDuration int // 目标总时长（秒），<=0 按镜头时间码累加兜底，仍<=0 默认 60 - MaxSegmentDur int // 单段最大时长（秒），<=0 默认 15 - MinSegmentDur int // 单段最小时长（秒），<=0 默认 5 - Refs Refs // 参考素材（角色/场景/道具/产品，具名） - Seed int64 // 随机种子基数，各段 = Seed + 段序号 - NegativePrompt string - ModelName string // 写入请求体 body.model - Schema map[string]any // 视频模型 schema（可选），nil 用默认通用结构 - MaxRefs int // 参考素材上限，<=0 默认 5 - MaxPromptChars int // prompt 截断长度（rune），<=0 不截断 - TokenConfig TokenConfig // 实体名替换 token 的生成配置 -} - -// TokenConfig token 前缀策略：可配置，默认全用 characterN（video-factory 兼容）。 -type TokenConfig struct { - // 是否按类别区分前缀。false=一律 characterN（推荐，模型只按 reference_urls 顺序识别）； - // true=character/scene/prop/product 分别编号。 - ByCategory bool - // 自定义前缀模板，如 "char%d"。为空时按 ByCategory 决定默认行为。 - Template string -} - -// Refs 参考素材，与 domain/plan 的 Refs 概念一致（自包含定义，不依赖 plan）。 -type Refs struct { - Characters []RefItem `json:"characters,omitempty"` - Scenes []RefItem `json:"scenes,omitempty"` - Props []RefItem `json:"props,omitempty"` - Products []RefItem `json:"products,omitempty"` -} - -type RefItem struct { - Name string `json:"name"` - URL string `json:"url"` -} - -// Segment 一个视频生成段：时间轴、段内镜头、prompt、参考素材、请求体。 -type Segment struct { - Index int // 段序号（从 0 起） - StartSec int // 段在全局时间轴上的起点（秒） - Duration int // 段时长（秒） - Shots []domain.Shot // 段内镜头（已回写对齐，不跨段） - Prompt string // 实体名→token 替换后的 prompt 文本 - Labels map[string]string // 实体名 → token（character1...） - RefURLs []string // 参考素材 URL，顺序与 token 对应 - Seed int64 - Request map[string]any // 完整视频 API 请求体（schema 嵌套） -} - -// Output Run 的产出。 -type Output struct { - Segments []Segment - TotalDuration int // 实际总时长（秒） -} -``` - -## 5. 阶段① 时间线构建（重建 + 回写对齐） - -入口：`BuildTimeline(in Input) (segShots [][]domain.Shot, segDurs []int, err error)` -（`segShots[i]` 为第 i 段的已对齐镜头列表，段数量与 `segDurs` 一致） - -### 5.1 rebuildDurations -丢弃 AI 的 `startTime/endTime`，按镜头文本量重建每段时长： -- 有文字（台词+旁白）的镜头：按语速（默认 4 字/秒，感叹/疑问多→6，低落→3）算最小时长 = `ceil(字数/语速) + 1`，低弹性； -- 纯视觉镜头：以 AI 原始时长意图为基准（`endTime-startTime`），高弹性； -- 盈余按弹性权重分配；超出目标则先压缩视觉镜头、再等比压缩有声镜头、最后截断尾部； -- 不变量：重建后各镜头时长之和 **精确等于** `TotalDuration`。 - -> 算法与 `video/domain/timeline.go` 的 `rebuildTimeline` 语义一致，但在 `pipeline` 内**自包含实现**（不 import plan）。 - -### 5.2 splitOversized -超 `MaxSegmentDur` 的镜头按句末断句（`。！？；\n!?.`）切子镜头，避免"说话说一半"；最后一片段 `≤ minSeg` 时减少拆分段数。重排 `index`。 - -### 5.3 calcSegmentDurations -将 `TotalDuration` 拆为多段： -- `numSegments = ceil(Total / MaxSegmentDur)`，`base = Total / num`，前 `Total % num` 段各 `+1`； -- 每段 `≥ MinSegmentDur`（不足时从后一向前借位）； -- 边界：`MinSegmentDur > MaxSegmentDur` 时收敛为相等。 - -### 5.4 alignToSegments（回写对齐，核心新增） -按段边界把镜头切分并对齐： -1. 以段边界 `segStart`/`segEnd`（秒）为切点； -2. 对每个与段窗口相交的镜头，裁剪到 `[max(shotStart, segStart), min(shotEnd, segEnd))`，时间码写回为段内绝对时间轴上的 `MM:SS`； -3. **跨边界镜头切成两个子镜头**，各留各段，剩余部分进入下一段的时间线； -4. **短残片并入**：切出的子镜头时长 `< 1s` 时并入相邻段（并入前一段末尾），避免产生超短镜头； -5. 不变量：每个镜头只属于一个段；段内镜头时间码落在该段窗口内；所有段首尾相接覆盖 `[0, TotalDuration)`。 - -输出：`segShots [][]domain.Shot`（每段一组已对齐镜头）+ `segDurs`，供阶段②逐段构建 prompt。 - -## 6. 阶段② Prompt 构建（模型语言·文本层） - -入口：`BuildSegmentPrompt(segShots []domain.Shot, refs Refs, tc TokenConfig, maxRefs int) (prompt string, labels map[string]string, refURLs []string)` - -1. **收集实体**：按 演员→场景→道具（→产品，若产品图固定携带）出现顺序收集段内具名实体，查 `refs` 拿 URL，上限 `MaxRefs`； -2. **生成 token**：按 `TokenConfig` 生成 `character1...`（默认，全用同一前缀、按引用顺序编号）或按类别前缀 `character1/scene1/prop1/product1`，或自定义模板； -3. **替换 prompt 文本**：把镜头文本中的实体名替换为 token；**按名字长度降序替换**，避免长名是短名前缀时的部分覆盖；无 URL 的实体保留原名； -4. **截断**：`MaxPromptChars > 0` 且超出时截断并追加 `...`（保留至少 50 字符下限）； -5. 返回 `prompt`（替换后）、`labels`（实体名→token）、`refURLs`（顺序与 token 对应）。 - -## 7. 阶段③ 请求体构建（模型语言·API 层） - -入口：`BuildVideoRequest(seg Segment, in Input) map[string]any` - -1. 组装 flat input： - ```go - map[string]any{ - "prompt": seg.Prompt, - "seed": seg.Seed, - // 以下有值才写 - "negative_prompt": in.NegativePrompt, - "duration": seg.Duration, - "media": buildMedia(seg.RefURLs, mediaTypeValue), // 参考图 media 数组 - "reference_urls": seg.RefURLs, - } - ``` -2. **schema 嵌套**：仿 video-factory `BuildSchemaRequest`——遍历 schema，无 `type` 键的节点视为分组递归进入，有 `type` 的字段从 flat input 取值、缺失填 `default`、可校验 required/type；`in.Schema == nil` 时用默认通用结构； -3. `body["model"] = in.ModelName`； -4. media 组装：参考图 URL 包装为 media 项，type 值取 `reference_image`（schema_mapping 可配置，复用路径解析思路）；媒体在请求中保持 URL 字符串（base64 转码属 I/O，不在本包做，由调用方处理）。 - -### 默认通用 schema（火山风格） -```json -{ - "body": { - "input": { - "prompt": { "type": "string", "required": true }, - "seed": { "type": "integer", "default": -1 }, - "negative_prompt": { "type": "string" }, - "duration": { "type": "integer" }, - "media": { "type": "array" }, - "reference_urls": { "type": "array" } - }, - "parameters": { - "model": { "type": "string" } - } - } -} -``` - -## 8. 入口 - -```go -// Run 编排①→②→③，纯函数，无 I/O。 -func Run(in Input) (*Output, error) -``` - -错误处理： -- `Shots` 为空 → 返回错误（不产出空请求）； -- `TotalDuration <= 0` 且镜头无法累加时长 → 兜底 60； -- schema 字段校验失败 → 返回明确错误。 - -## 9. 测试计划 - -| 用例 | 断言 | -|------|------| -| 时间线重建恒等 | 重建后镜头时长之和 == TotalDuration | -| 跨边界切分回写 | 段边界处的镜头被切成子镜头，各镜头时间码落在所属段窗口内，不跨段 | -| 短残片并入 | 切出的 `<1s` 子镜头并入相邻段 | -| token 替换 | 实体名替换为 character1...；长名优先替换；无 URL 实体保留原名 | -| token 可配置 | ByCategory=true 时按类别前缀；Template 生效 | -| schema 嵌套/默认值 | flat input 正确嵌套进 `{body:{input,parameters}}`，缺失字段填 default | -| 用户样本 golden test | 用 2026-08-11 实测的 3 镜头中文样本（时间码 00:00-00:07 等）跑通 Run，断言段数量、prompt 含 characterN、请求体结构 | - -## 10. 与现有链路的关系（本阶段不实现） - -- 新包完全独立，不接 workflow； -- 将来接入：写薄适配层把 `Segment.Request` 交给模型网关 `modelCall`（对应 `video/plan` 的 `SegmentParamsToMap` 语义），媒体 URL 转 base64 在适配层做； -- 现有 `split_shots`/`generate_segments_serial` 保持不动，两者可并存。 - -## 11. 评审记录 - -- 2026-08-11：方案 B（完全独立从零写）获用户确认；token 前缀=可配置（默认 characterN）；跨边界=切分+短残片并入；目标 API=通用抽象；挂载层=独立纯函数包；输入=外部直接传 shots。 \ No newline at end of file diff --git a/go.mod b/go.mod index 8e91b0c..988fc37 100644 --- a/go.mod +++ b/go.mod @@ -3,26 +3,18 @@ module ai-agent go 1.26.0 require ( - gitea.redpowerfuture.com/red-future/common v0.0.24 - github.com/bjang03/gmq v0.0.1 + gitea.redpowerfuture.com/red-future/common v0.0.31 + github.com/bjang03/gmq v0.0.2 github.com/cloudwego/eino v0.9.5 github.com/cloudwego/eino-examples v0.0.0-20260611092511-bd64846fbc1d github.com/cloudwego/eino-ext/components/model/qwen v0.1.9 github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2 github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 - github.com/gogf/gf/contrib/registry/consul/v2 v2.9.5 github.com/gogf/gf/v2 v2.10.2 github.com/google/uuid v1.6.0 - github.com/stretchr/testify v1.11.1 github.com/tidwall/gjson v1.18.0 - github.com/tidwall/sjson v1.2.5 - go.opentelemetry.io/otel/trace v1.44.0 ) -replace gitea.redpowerfuture.com/red-future/common v0.0.24 => ../common - -replace github.com/bjang03/gmq v0.0.1 => ../gmq - require ( github.com/BurntSushi/toml v1.5.0 // indirect github.com/armon/go-metrics v0.4.1 // indirect @@ -54,6 +46,7 @@ require ( github.com/go-playground/locales v0.14.1 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.30.1 // indirect + github.com/gogf/gf/contrib/registry/consul/v2 v2.9.5 // indirect github.com/gogf/gf/contrib/trace/otlphttp/v2 v2.9.5 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang-jwt/jwt/v5 v5.3.1 // indirect @@ -109,6 +102,7 @@ require ( github.com/spf13/cast v1.10.0 // indirect github.com/tidwall/match v1.1.1 // indirect github.com/tidwall/pretty v1.2.1 // indirect + github.com/tidwall/sjson v1.2.5 // indirect github.com/tiger1103/gfast-token v1.0.10 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/vcaesar/cedar v0.30.0 // indirect @@ -123,6 +117,7 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/sdk v1.38.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.7.1 // indirect go.uber.org/atomic v1.11.0 // indirect golang.org/x/arch v0.19.0 // indirect diff --git a/go.sum b/go.sum index 7b8043a..9141d8b 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,8 @@ cel.dev/expr v0.24.0/go.mod h1:hLPLo1W4QUmuYdA72RBX06QTs6MXw941piREPl3Yfiw= cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= cloud.google.com/go/compute/metadata v0.7.0/go.mod h1:j5MvL9PprKL39t166CoB1uVHfQMs4tFQZZcKwksXUjo= +gitea.redpowerfuture.com/red-future/common v0.0.31 h1:9H8nL5Drazcv7Hs9d4j+cXhaB+7uOllIUqEOyZy1Eao= +gitea.redpowerfuture.com/red-future/common v0.0.31/go.mod h1:xPU7aaMxn8rtNnWc2LDUXZL+IkaUkpQeLgflqw9FvdU= github.com/Azure/azure-sdk-for-go/sdk/azcore v1.17.0/go.mod h1:XCW7KnZet0Opnr7HccfUw1PLc4CjHqpcaxW8DHklNkQ= github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0/go.mod h1:iZDifYGJTIgIIkYRNWPENUnqx6bJ2xnSDFI2tjwZNuY= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= @@ -38,6 +40,8 @@ github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+Ce github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA= +github.com/bjang03/gmq v0.0.2 h1:3CcVorDXYoRIN65bbzwRuUxzkBCkEpHWmKHOkfXzUo0= +github.com/bjang03/gmq v0.0.2/go.mod h1:Y7TwWGuV4Cw97WUDaM7x+NC4kyFx1z44WAvNwJV3HV8= github.com/bluele/gcache v0.0.2/go.mod h1:m15KV+ECjptwSPxKhOhQoAFQVtUFjTVkc3H8o0t/fp0= github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= diff --git a/main.go b/main.go index 3ff9b1e..2113ee3 100644 --- a/main.go +++ b/main.go @@ -13,7 +13,9 @@ import ( sessionController "ai-agent/workflow/controller/session" workflowSkillController "ai-agent/workflow/controller/skill" toolController "ai-agent/workflow/controller/tool" + _ "ai-agent/workflow/service/flow/processor/builtin/media" _ "ai-agent/workflow/service/flow/processor/builtin/split_batch" + _ "ai-agent/workflow/service/flow/processor/builtin/split_segment" _ "ai-agent/workflow/service/flow/processor/builtin/split_shots_pipeline" "context" "os" diff --git a/tools/runner/react_agent.go b/tools/runner/react_agent.go index f700635..baa6afb 100644 --- a/tools/runner/react_agent.go +++ b/tools/runner/react_agent.go @@ -20,6 +20,7 @@ import ( type ReActEventType string const ( + ReActEventRoundStart ReActEventType = "round_start" ReActEventModelCall ReActEventType = "model_call" // 模型思考（step 开始） ReActEventToolCall ReActEventType = "tool_call" // 模型请求调用工具 ReActEventToolResult ReActEventType = "tool_result" // 工具返回结果 @@ -32,6 +33,7 @@ const ( // ReActEvent ReAct 过程事件，按发生顺序回调 type ReActEvent struct { Type ReActEventType + Id int64 Step int MaxStep int Description string // 工具用途说明（tool_call/tool_result 推给前端展示，不暴露工具名/参数） @@ -140,7 +142,6 @@ func (a *ReActAgent) Run(ctx context.Context, userInput string) (string, error) } // 无工具调用 → 最终回答 if len(toolCalls) == 0 { - a.emit(ReActEvent{Type: ReActEventAnswer, Answer: content}) return content, nil } messages["assistant_prompt"] = content diff --git a/update.sql b/update.sql index 6b8df15..690e849 100644 --- a/update.sql +++ b/update.sql @@ -725,7 +725,8 @@ CREATE TABLE IF NOT EXISTS black_deacon_exec_chat ( result_file_url VARCHAR(512) DEFAULT '', -- 结果文件路径（OSS） total_tokens INTEGER NOT NULL DEFAULT 0, -- 总token消耗 total_fee DOUBLE PRECISION NOT NULL DEFAULT 0, -- 总费用 - error_message TEXT DEFAULT '' -- 错误信息 + error_message TEXT DEFAULT '', -- 错误信息（友好提示） + error TEXT DEFAULT '' -- 错误明细（原始错误） ); -- 索引（高频查询） @@ -748,7 +749,10 @@ COMMENT ON COLUMN black_deacon_exec_chat.request_params IS '请求参数'; COMMENT ON COLUMN black_deacon_exec_chat.result_file_url IS '结果文件路径（OSS）'; COMMENT ON COLUMN black_deacon_exec_chat.total_tokens IS '总token消耗'; COMMENT ON COLUMN black_deacon_exec_chat.total_fee IS '总费用'; -COMMENT ON COLUMN black_deacon_exec_chat.error_message IS '错误信息'; +COMMENT ON COLUMN black_deacon_exec_chat.error_message IS '错误信息（友好提示）'; +COMMENT ON COLUMN black_deacon_exec_chat.error IS '错误明细（原始错误）'; +-- 兼容已有库：错误信息拆分，error_message 存友好提示、error 存原始错误 +ALTER TABLE black_deacon_exec_chat ADD COLUMN IF NOT EXISTS error TEXT DEFAULT ''; --------------------pgsql创建black_deacon_exec_chat表语句--------------------------- --------------------pgsql创建black_deacon_exec_workflow表语句--------------------------- @@ -771,7 +775,8 @@ CREATE TABLE IF NOT EXISTS black_deacon_exec_workflow ( status SMALLINT NOT NULL DEFAULT 1, -- 状态:1-运行中,2-成功,3-失败 total_tokens INTEGER NOT NULL DEFAULT 0, -- 总token消耗 total_fee DOUBLE PRECISION NOT NULL DEFAULT 0, -- 总费用 - error_message TEXT DEFAULT '' -- 错误信息 + error_message TEXT DEFAULT '', -- 错误信息（友好提示） + error TEXT DEFAULT '' -- 错误明细（原始错误） ); -- 索引（高频查询） @@ -797,7 +802,10 @@ COMMENT ON COLUMN black_deacon_exec_workflow.request_params IS '请求参数'; COMMENT ON COLUMN black_deacon_exec_workflow.status IS '状态:1-运行中,2-成功,3-失败'; COMMENT ON COLUMN black_deacon_exec_workflow.total_tokens IS '总token消耗'; COMMENT ON COLUMN black_deacon_exec_workflow.total_fee IS '总费用'; -COMMENT ON COLUMN black_deacon_exec_workflow.error_message IS '错误信息'; +COMMENT ON COLUMN black_deacon_exec_workflow.error_message IS '错误信息（友好提示）'; +COMMENT ON COLUMN black_deacon_exec_workflow.error IS '错误明细（原始错误）'; +-- 兼容已有库：错误信息拆分，error_message 存友好提示、error 存原始错误 +ALTER TABLE black_deacon_exec_workflow ADD COLUMN IF NOT EXISTS error TEXT DEFAULT ''; --------------------pgsql创建black_deacon_exec_workflow表语句--------------------------- --------------------pgsql创建black_deacon_exec_workflow_result表语句--------------------------- diff --git a/workflow/consts/node/node_template.go b/workflow/consts/node/node_template.go index 95a7032..052cae5 100644 --- a/workflow/consts/node/node_template.go +++ b/workflow/consts/node/node_template.go @@ -37,9 +37,11 @@ type NodeTypeMeta struct { Group NodeGroup `json:"group"` // 所属分组 Sort int `json:"sort"` // UI排序字段 Desc string `json:"desc,omitempty"` // 可选：节点简介 + PatchLayout bool `json:"patchLayout"` + IsMultiParameter bool `json:"isMultiParameter"` BatchExecOption bool `json:"batchExecOption"` - PreToolOption bool `json:"preToolOption"` - PostToolOption bool `json:"postToolOption"` + PreToolOption []NodePresetField `json:"preToolOption"` + PostToolOption []NodePresetField `json:"postToolOption"` IsSaveFileOption bool `json:"isSaveFileOption"` FormConfigOption bool `json:"formConfigOption"` ModelConfigOption bool `json:"modelConfigOption"` @@ -47,11 +49,17 @@ type NodeTypeMeta struct { PromptOption bool `json:"promptOption"` NegativePromptOption bool `json:"negativePromptOption"` PresetOption []NodePresetField `json:"presetOption"` - OutputField []string `json:"outputField"` + OutputField []NodeOutputField `json:"outputField"` +} + +type NodeOutputField struct { + Field string `json:"field"` + Label string `json:"label"` } type NodePresetField struct { Value string `json:"value"` + ValueSource []ValueSource `json:"valueSource"` Field string `json:"field"` Label string `json:"label"` Type string `json:"type"` @@ -61,6 +69,12 @@ type NodePresetField struct { Options []SelectOption `json:"options"` } +type ValueSource struct { + NodeId string `json:"nodeId"` + Field string `json:"field"` + Label string `json:"label"` +} + // FieldConstraint 字段约束 type FieldConstraint struct { // 数字类型：int、float、double、string @@ -91,20 +105,38 @@ var NodeGroupMetaList = []NodeGroupMeta{ var NodeTypeMetaList = []NodeTypeMeta{ { - Key: NodeTypeModel, - Name: "模型", - Group: NodeGroupBase, - Sort: 1, - Desc: "模型调用节点，可配置模型参数、模型配置、技能、提示语、结果汇集、结果保存、结果返回、结果展示等信息。", - BatchExecOption: true, - PreToolOption: true, - PostToolOption: true, + Key: NodeTypeModel, + Name: "模型", + Group: NodeGroupBase, + Sort: 1, + Desc: "模型调用节点，可配置模型参数、模型配置、技能、提示语、结果汇集、结果保存、结果返回、结果展示等信息。", + PatchLayout: true, + IsMultiParameter: true, + BatchExecOption: true, + PreToolOption: []NodePresetField{ + { + Field: "perTool", + Label: "前置方法", + Type: "select", + Required: false, + Options: []SelectOption{}, + }, + }, + PostToolOption: []NodePresetField{ + { + Field: "postTool", + Label: "后置方法", + Type: "select", + Required: false, + Options: []SelectOption{}, + }, + }, IsSaveFileOption: true, FormConfigOption: false, ModelConfigOption: true, SkillOption: false, PromptOption: true, - NegativePromptOption: true, + NegativePromptOption: false, }, { Key: NodeTypeDataMerge, @@ -112,9 +144,8 @@ var NodeTypeMetaList = []NodeTypeMeta{ Group: NodeGroupBase, Sort: 2, Desc: "结果汇集节点，可配置结果汇集方式、结果保存、结果返回、结果展示等信息。", + IsMultiParameter: false, BatchExecOption: false, - PreToolOption: false, - PostToolOption: false, IsSaveFileOption: false, FormConfigOption: false, ModelConfigOption: false, @@ -128,9 +159,8 @@ var NodeTypeMetaList = []NodeTypeMeta{ Group: NodeGroupBase, Sort: 3, Desc: "表单节点，可配置表单字段、表单配置、结果汇集、结果保存、结果返回、结果展示等信息。", + IsMultiParameter: false, BatchExecOption: false, - PreToolOption: false, - PostToolOption: false, IsSaveFileOption: false, FormConfigOption: true, ModelConfigOption: false, @@ -166,9 +196,7 @@ var NodeTypeMetaList = []NodeTypeMeta{ Sort: 5, Desc: "HTTP(S)接口节点，可配置HTTP(S)接口地址、请求方式、请求头、请求体、结果返回结构、结果返回方式、结果返回结构、结果返回方式等信息。", BatchExecOption: false, - PreToolOption: true, - PostToolOption: true, - IsSaveFileOption: false, + IsSaveFileOption: true, FormConfigOption: false, ModelConfigOption: false, SkillOption: false, @@ -305,9 +333,8 @@ var NodeTypeMetaList = []NodeTypeMeta{ Group: NodeGroupBase, Sort: 6, Desc: "脚本转写节点，把文案/视频分析结果通过大模型转写为结构化分镜脚本（[]Shot），作为视频生成节点的 shots 输入。", + IsMultiParameter: false, BatchExecOption: false, - PreToolOption: false, - PostToolOption: false, IsSaveFileOption: false, FormConfigOption: false, ModelConfigOption: true, @@ -334,7 +361,24 @@ var NodeTypeMetaList = []NodeTypeMeta{ Options: []SelectOption{}, }, }, - OutputField: []string{"prompt", "duration", "seed", "negative_prompt", "reference_urls"}, + OutputField: []NodeOutputField{ + { + Field: "prompt", + Label: "转写内容", + }, + { + Field: "duration", + Label: "时长", + }, + { + Field: "seed", + Label: "种子", + }, + { + Field: "reference_urls", + Label: "参考链接", + }, + }, }, { Key: NodeTypeSystemSum, diff --git a/workflow/controller/session/session_controller.go b/workflow/controller/session/session_controller.go index e00efb6..e48f51b 100644 --- a/workflow/controller/session/session_controller.go +++ b/workflow/controller/session/session_controller.go @@ -54,3 +54,8 @@ func (c *session) DeleteRecord(ctx context.Context, req *sessionDto.DeleteSessio func (c *session) ResultList(ctx context.Context, req *sessionDto.ListWorkflowResultReq) (res *flowDto.ListFlowExecutionTreeRes, err error) { return sessionService.SessionService.ResultList(ctx, req) } + +func (c *session) ResultDelete(ctx context.Context, req *sessionDto.DeleteWorkflowResultReq) (res *beans.ResponseEmpty, err error) { + err = sessionService.SessionService.ResultDelete(ctx, req) + return +} diff --git a/workflow/dao/session/exec_chat_dao.go b/workflow/dao/session/exec_chat_dao.go index e3822e1..9883b22 100644 --- a/workflow/dao/session/exec_chat_dao.go +++ b/workflow/dao/session/exec_chat_dao.go @@ -27,6 +27,14 @@ func (d *execChatDao) Insert(ctx context.Context, req *sessionDto.CreateExecChat return r.LastInsertId() } +func (d *execChatDao) Update(ctx context.Context, req *sessionDto.UpdateExecChatReq) (rows int64, err error) { + r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecChat).OmitEmpty().Data(&req).Where(entity.ExecChatCol.Id, req.Id).Update() + if err != nil { + return + } + return r.RowsAffected() +} + func (d *execChatDao) Delete(ctx context.Context, req *sessionDto.DeleteExecChatReq) (rows int64, err error) { r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecChat).Where(entity.ExecChatCol.Id, req.Id).Delete() if err != nil { diff --git a/workflow/dao/session/exec_workflow_dao.go b/workflow/dao/session/exec_workflow_dao.go index dc54408..702abb7 100644 --- a/workflow/dao/session/exec_workflow_dao.go +++ b/workflow/dao/session/exec_workflow_dao.go @@ -69,6 +69,24 @@ func (d *execWorkflowDao) List(ctx context.Context, creator string, page *beans. return } +// GetLatestBySessionAndFlow 查询会话+工作流下最近一次执行记录（按创建时间倒序，无记录返回 nil） +func (d *execWorkflowDao) GetLatestBySessionAndFlow(ctx context.Context, sessionId string, flowId int64) (res *entity.ExecWorkflow, err error) { + r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). + Where(entity.ExecWorkflowCol.SessionId, sessionId). + Where(entity.ExecWorkflowCol.FlowId, flowId). + OrderDesc(entity.ExecWorkflowCol.CreatedAt). + Limit(1). + One() + if err != nil { + return + } + if r.IsEmpty() { + return nil, nil + } + err = r.Struct(&res) + return +} + // ListBySession 查询会话下工作流执行记录（按创建时间倒序） func (d *execWorkflowDao) ListBySession(ctx context.Context, sessionId string) (res []*entity.ExecWorkflow, err error) { r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). @@ -84,22 +102,26 @@ func (d *execWorkflowDao) ListBySession(ctx context.Context, sessionId string) ( // ListDates 按创建人查询去重后的创建日期（倒序，支持分页；page 为 nil 返回全部日期） func (d *execWorkflowDao) ListDates(ctx context.Context, creator string, page *beans.Page) (dates []string, err error) { + fieldAlias := "create_date" m := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). - Fields("DATE(created_at) AS create_date"). + Fields("DATE("+entity.ExecWorkflowCol.CreatedAt+") AS "+fieldAlias). Where(entity.ExecWorkflowCol.Creator, creator). - Group("create_date"). - OrderDesc("create_date") + Group("DATE(" + entity.ExecWorkflowCol.CreatedAt + ")"). // 和select表达式保持一致，按自然日分组去重 + OrderDesc(fieldAlias) // 按日期别名倒序 + if page != nil { m.Page(int(page.PageNum), int(page.PageSize)) } + r, err := m.All() if err != nil { - return + return nil, err } + for _, rec := range r { - dates = append(dates, rec["create_date"].String()) + dates = append(dates, rec[fieldAlias].String()) } - return + return dates, nil } // ListByDates 按创建人查询指定创建日期（DATE(created_at) 命中）内的执行记录，按创建时间倒序 diff --git a/workflow/model/dto/flow/flow_execution_dto.go b/workflow/model/dto/flow/flow_execution_dto.go index a8b8746..98f4eb1 100644 --- a/workflow/model/dto/flow/flow_execution_dto.go +++ b/workflow/model/dto/flow/flow_execution_dto.go @@ -41,6 +41,8 @@ type ExecutedNode struct { Status node.NodeExecutionStatus `json:"status"` // 执行状态：成功/失败 } +//======================================================================================= + //============================================================================= // 原始入参结构体 diff --git a/workflow/model/dto/session/exec_chat_dto.go b/workflow/model/dto/session/exec_chat_dto.go index 60d747e..dc6b3f5 100644 --- a/workflow/model/dto/session/exec_chat_dto.go +++ b/workflow/model/dto/session/exec_chat_dto.go @@ -6,12 +6,17 @@ import ( type CreateExecChatReq struct { SessionId string `json:"sessionId" description:"所属会话ID"` - Duration int64 `json:"duration" description:"执行时长（秒）"` RequestParams entity.ExecChatRequestParams `json:"requestParams" description:"请求参数"` - ResultFileUrl string `json:"resultFileUrl" description:"结果文件路径"` - TotalTokens int `json:"totalTokens" description:"总token消耗"` - TotalFee float64 `json:"totalFee" description:"总费用"` - ErrorMessage string `json:"errorMessage" description:"错误信息"` +} + +type UpdateExecChatReq struct { + Id int64 `json:"id" v:"required#会话执行记录ID不能为空"` + Duration int64 `json:"duration" description:"执行时长（秒）"` + ResultFileUrl string `json:"resultFileUrl" description:"结果文件路径"` + TotalTokens int `json:"totalTokens" description:"总token消耗"` + TotalFee float64 `json:"totalFee" description:"总费用"` + ErrorMessage string `json:"errorMessage" description:"错误信息（友好提示）"` + Error string `json:"error" description:"错误明细（原始错误）"` } type DeleteExecChatReq struct { diff --git a/workflow/model/dto/session/exec_workflow_dto.go b/workflow/model/dto/session/exec_workflow_dto.go index f5886f2..98557fc 100644 --- a/workflow/model/dto/session/exec_workflow_dto.go +++ b/workflow/model/dto/session/exec_workflow_dto.go @@ -11,7 +11,8 @@ type CreateWorkflowReq struct { NodeGroupId string `json:"nodeGroupId" description:"节点组ID"` Status flow.FlowExecutionStatus `json:"status" description:"状态:1-运行中,2-成功,3-失败"` RequestParams *entity.FlowInfo `json:"requestParams" description:"请求参数"` - ErrorMessage string `json:"errorMessage" description:"错误信息"` + ErrorMessage string `json:"errorMessage" description:"错误信息（友好提示）"` + Error string `json:"error" description:"错误明细（原始错误）"` } type DeleteExecWorkflowReq struct { @@ -25,5 +26,6 @@ type UpdateWorkflowReq struct { Duration int64 `json:"duration" description:"执行时长（秒）"` TotalTokens int `json:"totalTokens" description:"总token消耗"` TotalFee float64 `json:"totalFee" description:"总费用"` - ErrorMessage string `json:"errorMessage" description:"错误信息"` + ErrorMessage string `json:"errorMessage" description:"错误信息（友好提示）"` + Error string `json:"error" description:"错误明细（原始错误）"` } diff --git a/workflow/model/dto/session/exec_workflow_result_dto.go b/workflow/model/dto/session/exec_workflow_result_dto.go index f72c71d..d98604d 100644 --- a/workflow/model/dto/session/exec_workflow_result_dto.go +++ b/workflow/model/dto/session/exec_workflow_result_dto.go @@ -6,7 +6,3 @@ type CreateWorkflowResultReq struct { ExecId int64 `json:"execId" description:"执行ID"` ResultFileUrl string `json:"resultFileUrl" description:"结果文件路径"` } - -type DeleteWorkflowResultReq struct { - Id int64 `json:"id" v:"required#ID不能为空"` -} diff --git a/workflow/model/dto/session/session_dto.go b/workflow/model/dto/session/session_dto.go index ea93b28..787dd68 100644 --- a/workflow/model/dto/session/session_dto.go +++ b/workflow/model/dto/session/session_dto.go @@ -15,6 +15,7 @@ type WebSocketConnectReq struct { } type WebSocketExecChatReq struct { + Id int64 `json:"id" dc:"id"` ModelId int64 `json:"modelId" dc:"模型ID" v:"required#模型ID不能为空"` Question string `json:"question" dc:"用户提问" v:"required#用户提问不能为空"` SystemPrompt string `json:"systemPrompt" dc:"系统提示词"` @@ -48,7 +49,7 @@ type VOSession struct { } type DeleteSessionReq struct { - g.Meta `path:"/delete" method:"post" tags:"会话管理" summary:"删除会话" dc:"删除会话"` + g.Meta `path:"/delete" method:"delete" tags:"会话管理" summary:"删除会话" dc:"删除会话"` SessionId string `json:"sessionId" v:"required#会话ID不能为空"` } @@ -83,7 +84,8 @@ type VOSessionInfoResult struct { ResultContent string `json:"resultContent" description:"结果文件内容（服务端已读取，前端直接展示）"` TotalTokens int `json:"totalTokens" dc:"总token消耗"` TotalFee float64 `json:"totalFee" dc:"总费用"` - ErrorMsg string `json:"errorMsg" dc:"错误信息"` + ErrorMsg string `json:"errorMsg" dc:"错误信息（友好提示）"` + Error string `json:"error" dc:"错误明细（原始错误）"` CreatedAt *gtime.Time `json:"createdAt" dc:"创建时间"` } @@ -91,3 +93,8 @@ type ListWorkflowResultReq struct { g.Meta `path:"/resultList" method:"get" tags:"会话管理" summary:"工作流执行结果树" dc:"按创建人分页查询工作流执行结果，按天分组返回树结构（日期→流程→结果文件），pageSize=每页天数，不传返回全部"` Page *beans.Page `json:"page"` } + +type DeleteWorkflowResultReq struct { + g.Meta `path:"/resultDelete" method:"delete" tags:"会话管理" summary:"删除工作流执行结果" dc:"删除工作流执行结果"` + Id int64 `json:"id" v:"required#ID不能为空"` +} diff --git a/workflow/model/entity/exec_chat.go b/workflow/model/entity/exec_chat.go index ad5343b..6d5d5e6 100644 --- a/workflow/model/entity/exec_chat.go +++ b/workflow/model/entity/exec_chat.go @@ -11,7 +11,8 @@ type ExecChat struct { ResultFileUrl string `orm:"result_file_url" json:"resultFileUrl" description:"结果文件路径"` TotalTokens int `orm:"total_tokens" json:"totalTokens" description:"总token消耗"` TotalFee float64 `orm:"total_fee" json:"totalFee" description:"总费用"` - ErrorMessage string `orm:"error_message" json:"errorMessage" description:"错误信息"` + ErrorMessage string `orm:"error_message" json:"errorMessage" description:"错误信息（友好提示）"` + Error string `orm:"error" json:"error" description:"错误明细（原始错误）"` } type ExecChatRequestParams struct { @@ -27,6 +28,7 @@ type execChatCol struct { TotalTokens string TotalFee string ErrorMessage string + Error string } var ExecChatCol = execChatCol{ @@ -38,4 +40,5 @@ var ExecChatCol = execChatCol{ TotalTokens: "total_tokens", TotalFee: "total_fee", ErrorMessage: "error_message", + Error: "error", } diff --git a/workflow/model/entity/exec_workflow.go b/workflow/model/entity/exec_workflow.go index 19c62ef..0c105dd 100644 --- a/workflow/model/entity/exec_workflow.go +++ b/workflow/model/entity/exec_workflow.go @@ -17,7 +17,8 @@ type ExecWorkflow struct { Status flow.FlowExecutionStatus `orm:"status" json:"status" description:"状态:1-运行中,2-成功,3-失败"` TotalTokens int `orm:"total_tokens" json:"totalTokens" description:"总token消耗"` TotalFee float64 `orm:"total_fee" json:"totalFee" description:"总费用"` - ErrorMessage string `orm:"error_message" json:"errorMessage" description:"错误信息"` + ErrorMessage string `orm:"error_message" json:"errorMessage" description:"错误信息（友好提示）"` + Error string `orm:"error" json:"error" description:"错误明细（原始错误）"` } type execWorkflowCol struct { @@ -31,6 +32,7 @@ type execWorkflowCol struct { TotalTokens string TotalFee string ErrorMessage string + Error string } var ExecWorkflowCol = execWorkflowCol{ @@ -44,4 +46,5 @@ var ExecWorkflowCol = execWorkflowCol{ TotalTokens: "total_tokens", TotalFee: "total_fee", ErrorMessage: "error_message", + Error: "error", } diff --git a/workflow/model/entity/flow_user.go b/workflow/model/entity/flow_user.go index 66bc276..758fce6 100644 --- a/workflow/model/entity/flow_user.go +++ b/workflow/model/entity/flow_user.go @@ -29,6 +29,8 @@ type FlowNode struct { OutputConfig []map[string]any `json:"outputConfig"` Prompt string `json:"prompt"` NegativePrompt string `json:"negativePrompt"` + PatchLayout bool `json:"patchLayout"` + Templates []map[string]any `json:"templates"` //SkillName string `json:"skillName"` //PromptContent string `json:"promptContent"` //InputSource []FlowNodeInputSource `json:"inputSource"` // 前端指定：来源节点ID @@ -37,33 +39,36 @@ type FlowNode struct { } type ModelItem struct { - ModelId int64 `json:"modelId,string"` - ModelName string `json:"modelName"` - ModelFormFields []map[string]any `json:"modelFormFields"` - ModelRequestParams map[string]any `json:"modelRequestParams"` - ModelResponseBodyMapping map[string]any `json:"modelResponseBodyMapping"` + ModelId int64 `json:"modelId,string"` + ModelName string `json:"modelName"` + ModelFormFields []map[string]any `json:"modelFormFields"` + ModelRequestParams map[string]any `json:"modelRequestParams"` + ModelRequestParamsPath []FlowModelParams `json:"modelRequestParamsPath"` + ModelResponseBodyMapping map[string]any `json:"modelResponseBodyMapping"` } -type FlowNodeInputSource struct { - NodeId string `json:"nodeId"` - QuoteOutput bool `json:"quoteOutput"` - Field []string `json:"field"` - FieldMap []FlowField `json:"fieldMap"` +type FlowModelParams struct { + Label string `json:"label"` + Path string `json:"path"` + Type string `json:"type"` + Required bool `json:"required"` + Value any `json:"value"` + ValueSource []ValueSource `json:"valueSource"` + RefsName string `json:"refsName"` } -type FlowField struct { - Key string `json:"key"` - Value string `json:"value"` - Desc string `json:"desc"` +type ValueSource struct { + NodeId string `json:"nodeId"` + Field string `json:"field"` + Label string `json:"label"` } // SubFlowConfig 子流程节点配置 type SubFlowConfig struct { - WorkflowId int64 `json:"workflowId"` - WorkflowName string `json:"workflowName"` - Fields []map[string]any `json:"fields"` - MaxConcurrency int `json:"maxConcurrency"` // 子流程并发数 - InputSource []FlowNodeInputSource `json:"inputSource"` // 前端指定：来源节点ID + WorkflowId int64 `json:"workflowId"` + WorkflowName string `json:"workflowName"` + Fields []map[string]any `json:"fields"` + MaxConcurrency int `json:"maxConcurrency"` // 子流程并发数 } type FlowEdge struct { diff --git a/workflow/service/flow/flow_checkpoint_store.go b/workflow/service/flow/flow_checkpoint_store.go index 31349b3..ed3929d 100644 --- a/workflow/service/flow/flow_checkpoint_store.go +++ b/workflow/service/flow/flow_checkpoint_store.go @@ -32,8 +32,6 @@ func init() { // ========== 4. entity 核心链路类型（递归自 *entity.FlowNode） ========== schema.RegisterName[*entity.FlowNode]("entity.FlowNode") schema.RegisterName[node.NodeType]("node.NodeType") - schema.RegisterName[entity.FlowNodeInputSource]("entity.FlowNodeInputSource") - schema.RegisterName[entity.FlowField]("entity.FlowField") schema.RegisterName[*entity.SubFlowConfig]("entity.SubFlowConfig") //schema.RegisterName[node.NodeFormField]("node.NodeFormField") schema.RegisterName[entity.ModelItem]("node.ModelItem") diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go index 8e48b92..905a999 100644 --- a/workflow/service/flow/flow_ws_exec.go +++ b/workflow/service/flow/flow_ws_exec.go @@ -7,7 +7,9 @@ import ( flowDto "ai-agent/workflow/model/dto/flow" sessionDto "ai-agent/workflow/model/dto/session" "ai-agent/workflow/model/entity" + "bytes" "context" + "encoding/json" "errors" "fmt" "sync" @@ -99,6 +101,7 @@ func (r *wsProgressReporter) ReportComplete(nodeId, nodeName string, nodeIndex, func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload interface{}) { execPayload := new(sessionDto.WebSocketExecWorkflowReq) if err := gconv.Struct(payload, execPayload); err != nil { + glog.Errorf(ctx, "工作流执行参数解析失败: %v", err) _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "执行参数解析失败", Error: err.Error()}) return } @@ -132,6 +135,7 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int flowName = flowUser.FlowName } if e := ensureSession(saveCtx, conn.SessionId, flowName); e != nil { + glog.Errorf(saveCtx, "工作流会话创建失败: %v", e) _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流会话创建失败", Error: fmt.Sprintf("%v", e)}) } @@ -142,7 +146,7 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", len(execPayload.FlowContent.Nodes))}) - execId, err := execute(progressCtx, conn.SessionId, execPayload) + execId, err := executeOrResume(progressCtx, conn, execPayload) recordWorkflow(saveCtx, execId, time.Since(start), err) if err != nil { _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: err.Error()}) @@ -163,14 +167,16 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int func recordWorkflow(ctx context.Context, id int64, duration time.Duration, runErr error) { // exec_workflow 状态沿用 1-运行中,2-成功,3-失败；前端结果卡片也只识别 1/2/3 // （4 会误显示为"运行中"），故取消同样记为失败，错误信息写"用户已终止执行" + // error_message 存友好提示，error 存原始错误明细 status := flow.FlowExecutionStatusSuccess - errorMessage := "" + var errorMessage, errorDetail string if runErr != nil { status = flow.FlowExecutionStatusFailed if errors.Is(runErr, context.Canceled) { errorMessage = errWorkflowTerminated } else { - errorMessage = runErr.Error() + errorMessage = "工作流执行失败" + errorDetail = runErr.Error() } } _, err := sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{ @@ -178,6 +184,7 @@ func recordWorkflow(ctx context.Context, id int64, duration time.Duration, runEr Status: status.Code(), Duration: int64(duration.Seconds()), ErrorMessage: errorMessage, + Error: errorDetail, }) if err != nil { glog.Errorf(ctx, "exec_workflow 落库失败: %v", err) @@ -222,11 +229,43 @@ func writeJSON(conn *wsCommon.WsConnection, data interface{}) error { return conn.WriteJSON(data) } +// executeOrResume 决策工作流执行方式： +// - 同会话+同工作流的最近一次执行失败，且本次传递参数与上次一致 → 断点续跑（reExecute，复用原执行记录，从失败断点继续） +// - 其余情况（上次成功 / 上次参数与本次不同 / 无历史记录 / 查询出错）→ 全新执行（execute） +func executeOrResume(ctx context.Context, conn *wsCommon.WsConnection, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { + lastExec, err := sessionDao.ExecWorkflowDao.GetLatestBySessionAndFlow(ctx, conn.SessionId, req.FlowId) + if err != nil { + glog.Errorf(ctx, "查询最近工作流执行记录失败: %v", err) + return execute(ctx, conn, req) + } + if lastExec != nil && *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() && flowContentEqual(lastExec.RequestParams, req.FlowContent) { + _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{ + "id": lastExec.Id, + }}) + return reExecute(ctx, lastExec.Id) + } + return execute(ctx, conn, req) +} + +// flowContentEqual 判断两次工作流参数是否一致（JSON 序列化后字节比对。 +// Go struct 按字段声明序序列化、map 键自动排序，同一内容结果确定，可用于参数等价判断） +func flowContentEqual(a, b *entity.FlowInfo) bool { + if a == nil || b == nil { + return a == b + } + ab, err1 := json.Marshal(a) + bb, err2 := json.Marshal(b) + if err1 != nil || err2 != nil { + return false + } + return bytes.Equal(ab, bb) +} + // execute 执行工作流（首次执行） -func execute(ctx context.Context, sessionId string, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { +func execute(ctx context.Context, conn *wsCommon.WsConnection, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) { var nodeGroupId = uuid.NewString() id, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{ - SessionId: sessionId, + SessionId: conn.SessionId, FlowId: req.FlowId, NodeGroupId: nodeGroupId, Status: flow.FlowExecutionStatusRunning.Code(), @@ -235,7 +274,10 @@ func execute(ctx context.Context, sessionId string, req *sessionDto.WebSocketExe if err != nil { return } - err = BuildExecution(ctx, true, req.FlowId, id, nodeGroupId, sessionId, req.FlowContent) + _ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{ + "id": id, + }}) + err = BuildExecution(ctx, true, req.FlowId, id, nodeGroupId, conn.SessionId, req.FlowContent) if err != nil { return } diff --git a/workflow/service/flow/lambda_node.go b/workflow/service/flow/lambda_node.go index 3b45def..bc17cba 100644 --- a/workflow/service/flow/lambda_node.go +++ b/workflow/service/flow/lambda_node.go @@ -3,6 +3,7 @@ package flow import ( "ai-agent/gateway" "ai-agent/workflow/consts/flow" + "ai-agent/workflow/consts/model" "ai-agent/workflow/consts/node" "ai-agent/workflow/consts/public" nodeDao "ai-agent/workflow/dao/node" @@ -12,6 +13,7 @@ import ( sessionDto "ai-agent/workflow/model/dto/session" "ai-agent/workflow/model/entity" "ai-agent/workflow/service/flow/processor" + "ai-agent/workflow/service/flow/processor/builtin/media" "context" "encoding/base64" "encoding/json" @@ -20,6 +22,7 @@ import ( "sync" "gitea.redpowerfuture.com/red-future/common/db/gfdb" + "gitea.redpowerfuture.com/red-future/common/utils" "github.com/cloudwego/eino-examples/compose/batch/batch" "github.com/cloudwego/eino/compose" "github.com/gogf/gf/v2/database/gdb" @@ -53,14 +56,16 @@ func ModelLambda(ctx context.Context, input any) (any, error) { return nil, fmt.Errorf("入参类型错误") } - // 1. 解析 valueSource 引用，填充模型请求参数 - ProcessValueSourceRecursive(nodeInput.Config.ModelConfig.ModelRequestParams, nodeInput.Global) - - // 1.5 剔除 value 为空的字段；数组/枚举元素整体为空时移除整个元素（0/false 视为有效值） - CleanEmptyModelParams(nodeInput.Config.ModelConfig.ModelRequestParams) + modelParams, err := BuildModelRequestBody(nodeInput.Config.ModelConfig.ModelRequestParamsPath, nodeInput.Global) + if err != nil { + return nil, err + } // 2. 前置工具：决定模型调用入参（单次/多次） - paramsList, err := invokePreTool(ctx, nodeInput.Config.PreTool, nodeInput.Config.ModelConfig.ModelRequestParams) + // 入参统一为扁平模型请求体（BuildModelRequestBody 输出，key 为点分路径）。 + // 分批处理器按默认上限拆分集合字段，其余前置工具（如 split_shots_pipeline）读取扁平参数。 + preToolParams := modelParams + paramsList, err := invokePreTool(ctx, nodeInput.Config.PreTool, preToolParams) if err != nil { return nil, err } @@ -69,21 +74,22 @@ func ModelLambda(ctx context.Context, input any) (any, error) { var outputRes []map[string]any var totalTokens int64 var totalCost float64 - if nodeInput.Config.IsBatchExec && len(paramsList) > 1 { + if len(paramsList) > 1 { // 异步批量执行：并发请求模型，等待全部返回后再继续，避免下游读到空结果 results := make([][]map[string]any, len(paramsList)) tokenRes := make([]*gateway.ModelCallRes, len(paramsList)) errs := make([]error, len(paramsList)) + isInference := make([]bool, len(paramsList)) var wg sync.WaitGroup for i, params := range paramsList { wg.Add(1) go func(i int, params map[string]any) { defer wg.Done() - results[i], tokenRes[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params) + results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt) }(i, params) } wg.Wait() - for i, res := range results { + for i := range results { if errs[i] != nil { return nil, errs[i] } @@ -91,11 +97,19 @@ func ModelLambda(ctx context.Context, input any) (any, error) { totalTokens += tokenRes[i].TotalTokens totalCost += tokenRes[i].Cost } - outputRes = append(outputRes, res...) + } + // 推理模型（分批同一模型，isInference 各批一致）：分批结果拼到单个字段（单条输出记录）； + // 非推理模型保持逐条展平 + if isInference[0] { + outputRes = mergeInferenceBatchResults(results) + } else { + for _, res := range results { + outputRes = append(outputRes, res...) + } } } else { for _, params := range paramsList { - res, modelRes, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params) + res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt) if err != nil { return nil, err } @@ -109,7 +123,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { // 3.5 把本次节点消耗的 token/费用写入节点执行记录，供汇总节点聚合到 exec_workflow if nodeInput.NodeExecutionId > 0 && (totalTokens > 0 || totalCost > 0) { - if _, err := nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{ + if _, err = nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{ Id: nodeInput.NodeExecutionId, TokenInfo: []map[string]any{{ "total_tokens": totalTokens, @@ -120,16 +134,61 @@ func ModelLambda(ctx context.Context, input any) (any, error) { } } - // 4. 后置工具：加工模型输出（透传原始请求参数，供后置工具读取合并配置等） - outputRes, err = invokePostTool(ctx, nodeInput.Config.PostTool, outputRes, nodeInput.Config.ModelConfig.ModelRequestParams) - if err != nil { - return nil, err + // 4.5 视频模型节点返回多个视频时，自动调用视频合成工具（concat_videos）合并为单条； + // 已显式配置 concat_videos 后置工具时跳过，避免重复合并 + if nodeInput.Config.PostTool != media.ProcessorName && len(outputRes) > 1 && isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId) { + outputRes, err = invokePostTool(ctx, media.ProcessorName, outputRes, map[string]any{"callback_url": "callback_url", "upload": true}) + if err != nil { + return nil, err + } + } else { + // 4. 后置工具：加工模型输出（透传原始请求参数，供后置工具读取合并配置等） + outputRes, err = invokePostTool(ctx, nodeInput.Config.PostTool, outputRes, modelParams) + if err != nil { + return nil, err + } } nodeInput.Config.OutputResult = outputRes return nodeInput, nil } +// isVideoModel 判断模型是否为视频模型（模型类型 TypeVideo=600），用于视频节点多视频自动合成判断 +func isVideoModel(ctx context.Context, modelId int64) bool { + modelInfo, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: modelId}) + if err != nil { + g.Log().Warningf(ctx, "查询模型配置失败，跳过自动视频合成 modelId=%d err=%v", modelId, err) + return false + } + return modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeVideo +} + +// mergeInferenceBatchResults 推理模型分批结果拼接为单条输出记录： +// 各批结果按批序对同名 key 的值做字符串拼接（"拼到一个字段"），最终返回单条 {key:值} 记录。 +// 非字符串值（如结构/数组字段）取最后一份，避免误拼接。 +func mergeInferenceBatchResults(results [][]map[string]any) []map[string]any { + merged := make(map[string]any) + for _, res := range results { + for _, record := range res { + for key, val := range record { + prev, has := merged[key] + if !has { + merged[key] = val + continue + } + sPrev, pOK := prev.(string) + sVal, vOK := val.(string) + if pOK && vOK { + merged[key] = sPrev + "\n" + sVal + continue + } + merged[key] = val + } + } + } + return []map[string]any{merged} +} + // invokePreTool 执行前置处理器，把模型请求参数转换为模型调用入参列表。 // 前置处理器契约：入参即模型请求参数本体；返回值： // - map[string]any 一次模型调用，入参为返回值 @@ -137,7 +196,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { // - nil 视为异常，节点失败（不允许静默跳过模型调用） func invokePreTool(ctx context.Context, processorName string, modelParams map[string]any) (paramsList []map[string]any, err error) { if processorName == "" { - return []map[string]any{modelParams}, nil + return []map[string]any{stripInternalKeys(modelParams)}, nil } data, err := processor.Call(ctx, processorName, modelParams) if err != nil { @@ -147,14 +206,32 @@ func invokePreTool(ctx context.Context, processorName string, modelParams map[st case nil: return nil, fmt.Errorf("前置处理器[%s]返回空", processorName) case map[string]any: - return []map[string]any{v}, nil + return []map[string]any{stripInternalKeys(v)}, nil case []map[string]any: - return v, nil + list := make([]map[string]any, 0, len(v)) + for _, m := range v { + list = append(list, stripInternalKeys(m)) + } + return list, nil default: return nil, fmt.Errorf("前置处理器[%s]返回类型不支持: %T", processorName, data) } } +// stripInternalKeys 剥离 __ 前缀的内部键（如 __segment_fields/__produced）， +// 模型网关做参数严格校验（CheckParams strictUnknown）会拒绝未知字段，内部标记不得随请求体下发。 +func stripInternalKeys(params map[string]any) map[string]any { + if params == nil { + return params + } + for k := range params { + if strings.HasPrefix(k, "__") { + delete(params, k) + } + } + return params +} + // invokePostTool 执行后置处理器，加工模型调用结果。 // 后置处理器契约：入参 {"output": 模型输出结果列表, "request": 原始模型请求参数}（列表须包成对象传入）；返回值： // - []map[string]any 替换模型输出 @@ -444,9 +521,10 @@ func SummaryLambda(ctx context.Context, input any) (any, error) { // collectSaveFileResults 按两层规则收集需入库的文件结果： // 第一层：节点须开启"保存文件"（IsSaveFile）； -// 第二层：key 取自模型响应内容 respBody（即节点 OutputResult 的各字段），命中 -// ModelResponseBodyMapping 才入库；原始响应体 key（respBody）恒入库（不要求映射声明）。 -// 结果值为 http(s) URL 直接使用；非路径值（base64 图片/文本）先上传 OSS 换取 URL， +// 第二层：key 取自节点 OutputResult 的各字段，命中 ModelResponseBodyMapping 才入库； +// 原始响应体 key（respBody）恒入库（不要求映射声明）；HTTP 节点产出以 http_file_url:{key} +// 标记的字段（IsSaveFile 时由 HttpCallResultLambda 生成）恒入库（无模型响应映射可查）。 +// 结果值为 http(s) URL 或 MinIO 对象裸路径直接使用；非路径值（base64 图片/文本）先上传 OSS 换取 URL， // 文本内容以 .inc 扩展名存储。 func collectSaveFileResults(ctx context.Context, execInput *flowDto.FlowExecutionInput) []*sessionDto.CreateWorkflowResultReq { if execInput == nil { @@ -458,13 +536,14 @@ func collectSaveFileResults(ctx context.Context, execInput *flowDto.FlowExecutio if nodeConfig == nil || len(nodeConfig.OutputResult) == 0 || !nodeConfig.IsSaveFile { continue } - // 第二层：key 取自模型响应内容 respBody（即节点 OutputResult 的各字段）， - // 命中 ModelResponseBodyMapping 才入库；原始响应体 key（respBody）恒入库 + // 第二层：key 取自节点 OutputResult 的各字段， + // 命中 ModelResponseBodyMapping 才入库；respBody 与 HTTP 节点 http_file_url:{key} 标记恒入库 saveKeys := nodeConfig.ModelConfig.ModelResponseBodyMapping for _, respBody := range nodeConfig.OutputResult { for key, val := range gconv.Map(respBody) { - if _, ok := saveKeys[key]; !ok { - if key != "respBody" { + isHTTPFile := strings.HasPrefix(key, "http_file_url:") + if !isHTTPFile { + if _, ok := saveKeys[key]; !ok && key != "respBody" { continue } } @@ -486,7 +565,7 @@ func collectSaveFileResults(ctx context.Context, execInput *flowDto.FlowExecutio } // resolveSaveFileResult 解析结果值为可入库的 URL： -// - 已是 http(s) URL → 直接返回 +// - 已是 http(s) URL 或 MinIO 对象裸路径 → 直接返回 // - 非路径（base64 图片/文本）→ 上传 OSS 换取 URL func resolveSaveFileResult(ctx context.Context, val any) (string, error) { isPath, path, fileBytes, ext := resolveFileContent(val) @@ -513,6 +592,10 @@ func resolveFileContent(val any) (isPath bool, path string, fileBytes []byte, ex if isFileURL(s) { return true, s, nil, "" } + // MinIO 对象裸路径（无 http 前缀，模型网关转存 OSS 后返回） + if utils.IsOSSPath(s) { + return true, s, nil, "" + } // data URI：data:;base64, if b, mime, ok := parseDataURI(s); ok { return false, "", b, extOfMime(mime) diff --git a/workflow/service/flow/lambda_node_util.go b/workflow/service/flow/lambda_node_util.go index 3686ba7..aef3001 100644 --- a/workflow/service/flow/lambda_node_util.go +++ b/workflow/service/flow/lambda_node_util.go @@ -2,10 +2,13 @@ package flow import ( "ai-agent/gateway" + "ai-agent/workflow/consts/model" "ai-agent/workflow/consts/node" flowDto "ai-agent/workflow/model/dto/flow" "context" "fmt" + "regexp" + "strings" "sync" commonHttp "gitea.redpowerfuture.com/red-future/common/http" @@ -57,20 +60,31 @@ func Notify(taskId string, result any) { delete(asyncTasks, taskId) } -// ModelCallResultLambda 调用模型并返回输出内容列表，同时回传本次调用的 token/费用（*gateway.ModelCallRes）， -// 供调用方（ModelLambda）累计写入节点执行记录 token_info，最后由汇总节点聚合到 exec_workflow。 -func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, modelRequestParams map[string]any) ([]map[string]any, *gateway.ModelCallRes, error) { +// ModelCallResultLambda 调用模型并返回输出内容列表，同时回传本次调用的 token/费用（*gateway.ModelCallRes） +// 与是否推理模型（供 ModelLambda 决定分批结果是否拼接），供调用方（ModelLambda）累计写入节点执行记录 +// token_info，最后由汇总节点聚合到 exec_workflow。 +func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, modelRequestParams map[string]any, prompt string) ([]map[string]any, *gateway.ModelCallRes, bool, error) { modelInfo, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: modelId}) if err != nil { - return nil, nil, fmt.Errorf("获取模型配置失败: %w", err) + return nil, nil, false, fmt.Errorf("获取模型配置失败: %w", err) } + businessParams := make(map[string]any) + if !g.IsEmpty(prompt) { + if modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeVideo { + businessParams["user_prompt"] = prompt + } else if modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeInference { + businessParams["system_prompt"] = prompt + } + } + // 推理模型：分批调用结果需拼接为单个字段，模型类型仅网关配置携带，此处顺带判断 + isInference := modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeInference // 异步模型 msgTopic 由 gateway.ModelCallResult 在为空时自动生成（唯一、带业务标识），调用方无需管理 - responseParams, err := gateway.ModelCallResult(ctx, modelId, modelInfo.ModelManage.ResponseType, sessionId, modelRequestParams, nil) + responseParams, err := gateway.ModelCallResult(ctx, modelId, modelInfo.ModelManage.ResponseType, sessionId, modelRequestParams, businessParams) if err != nil { - return nil, nil, err + return nil, nil, false, err } if g.IsEmpty(responseParams) { - return nil, nil, fmt.Errorf("生成内容为空") + return nil, nil, false, fmt.Errorf("生成内容为空") } outputRes := make([]map[string]any, 0) for key, val := range responseParams.Content { @@ -78,7 +92,7 @@ func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, key: val, }) } - return outputRes, responseParams, nil + return outputRes, responseParams, isInference, nil } func HttpCallResultLambda(ctx context.Context, nodeInput *flowDto.NodeExecutionInput) ([]map[string]any, error) { @@ -137,6 +151,9 @@ func HttpCallResultLambda(ctx context.Context, nodeInput *flowDto.NodeExecutionI // 递归剥掉 {type, value/attrs} 包裹层，只保留 key/value wrapper := UnwrapSchemaWrapper(body) newBody := gconv.Map(wrapper) + // body 值若为 MinIO 裸路径（模型网关转存 OSS 后返回，无 http 前缀）， + // 补上前缀供目标 HTTP 服务直接下载文件 + addFilePathPrefix(ctx, url, newBody) // 1. 自己生成唯一 taskId（不用前端给） taskId := "my_task_" + uuid.New().String() // 自己生成唯一ID @@ -209,3 +226,155 @@ func HttpCallResultLambda(ctx context.Context, nodeInput *flowDto.NodeExecutionI return outputRes, nil } + +// addFilePathPrefix 递归把 body 中的 MinIO 裸路径（模型网关转存 OSS 后返回，无 http 前缀）补上文件前缀， +// 供目标 HTTP 服务直接下载文件；已是完整 URL 的值保持不变 +func addFilePathPrefix(ctx context.Context, url string, body map[string]any) { + prefix, err := utils.GetFileAddressPrefix(ctx) + if err != nil { + g.Log().Warningf(ctx, "获取文件前缀失败，保持原路径: %v", err) + return + } + for k, v := range body { + body[k] = prependFilePathPrefix(prefix, v) + } + // template/template 模板接口要求 video_urls 为数组：标量值包装为单元素数组 + if strings.Contains(url, "template/template") { + if v, ok := body["video_urls"]; ok { + body["video_urls"] = toVideoURLsArray(v) + } + if v, ok := body["subtitles"]; ok { + a := new([]flowDto.Sentence) + err = gconv.Structs(v, a) + v, err = BuildSubtitles(a) + body["subtitles"] = v + } + } +} + +// toVideoURLsArray 把标量 video_urls 包装为数组；已是数组/切片则原样保留 +func toVideoURLsArray(v any) any { + switch val := v.(type) { + case string: + if val == "" { + return []string{} + } + return []string{val} + case []string, []any: + return val + default: + return v + } +} + +// prependFilePathPrefix 对单个值加前缀，递归处理嵌套 map/切片 +func prependFilePathPrefix(prefix string, v any) any { + switch val := v.(type) { + case string: + if utils.IsOSSPath(val) { + return prefix + val + } + return val + case map[string]any: + for k, item := range val { + val[k] = prependFilePathPrefix(prefix, item) + } + return val + case []any: + for i, item := range val { + val[i] = prependFilePathPrefix(prefix, item) + } + return val + case []map[string]any: + for _, m := range val { + for k, item := range m { + m[k] = prependFilePathPrefix(prefix, item) + } + } + return val + default: + return val + } +} + +// BuildSubtitles 核心工具：单个sentence生成多条subtitle +func BuildSubtitles(sents *[]flowDto.Sentence) ([]flowDto.Subtitle, error) { + var subtitles []flowDto.Subtitle + + for _, sent := range *sents { + // 1. 先按标点把文本拆成多个片段（保留标点） + segList := splitTextByPunct(sent.Text) + if len(segList) == 0 { + continue + } + + wordIdx := 0 + allWords := sent.Words + // 2. 遍历每个文本片段，匹配对应的Words + for _, seg := range segList { + // 去除文本片段的标点，方便和Word.Word拼接内容匹配 + segClean := strings.ReplaceAll(seg, "，", "") + segClean = strings.ReplaceAll(segClean, "。", "") + segClean = strings.ReplaceAll(segClean, "；", "") + segClean = strings.ReplaceAll(segClean, "！", "") + segClean = strings.ReplaceAll(segClean, "？", "") + + var collectWords []flowDto.Word + var currentText strings.Builder + + // 收集Word直到拼接内容覆盖当前分段 + for wordIdx < len(allWords) { + word := allWords[wordIdx] + currentText.WriteString(word.Word) + collectWords = append(collectWords, word) + wordIdx++ + + // 当拼接的文本包含当前分段的纯文本时，停止收集 + if strings.Contains(currentText.String(), segClean) { + break + } + } + + if len(collectWords) == 0 { + continue + } + + // 3. 生成字幕（时间戳取首尾Word的时间） + sub := flowDto.Subtitle{ + Start: collectWords[0].StartTime, + End: collectWords[len(collectWords)-1].EndTime, + Text: segClean, + } + subtitles = append(subtitles, sub) + } + } + + return subtitles, nil +} + +// splitTextByPunct 按中文标点分割句子，同时保留标点在分段内 +// 例如："这个叫高血压调理方，注意是根源调理不是临时缓解，" +// 会变成：["这个叫高血压调理方，", "注意是根源调理不是临时缓解，"] +func splitTextByPunct(raw string) []string { + // 匹配中文标点并保留在文本中，按标点位置切分 + re := regexp.MustCompile(`[，。；！？]`) + // 先找到所有标点的位置 + indexes := re.FindAllStringIndex(raw, -1) + if len(indexes) == 0 { + return []string{raw} + } + + var res []string + prev := 0 + for _, idx := range indexes { + end := idx[1] // 标点的结束位置 + seg := raw[prev:end] + res = append(res, seg) + prev = end + } + // 处理最后一段没有标点的文本 + if prev < len(raw) { + res = append(res, raw[prev:]) + } + return res +} diff --git a/workflow/service/flow/lambda_script_transcribe.go b/workflow/service/flow/lambda_script_transcribe.go index 8ad5c45..d5aee8e 100644 --- a/workflow/service/flow/lambda_script_transcribe.go +++ b/workflow/service/flow/lambda_script_transcribe.go @@ -35,8 +35,9 @@ const defaultScriptTranscribeSystemPrompt = `你是短剧分镜脚本师。请 时间码需前后衔接、覆盖整个内容时长。直接输出 JSON 数组，不要输出其他文字。` // ScriptTranscribeLambda 脚本转写节点： -// 把节点输入（文案/视频分析结果，经 valueSource 解析）通过大模型转写为固定结构的 []domain.Shot， -// 产出为 [{"shots": [...]}]，供视频生成节点的 ModelRequestParams.shots 引用。 +// 把节点输入（文案/视频分析结果，经 valueSource 解析）通过大模型转写为 []pipeline.Shot， +// 再经 split_shots_pipeline 前置处理器拆成各段扁平请求参数列表（[{"prompt","duration","seed",...},...]）， +// 供下游视频生成节点逐段引用聚合。 func ScriptTranscribeLambda(ctx context.Context, input any) (any, error) { nodeInput, ok := input.(*flowDto.NodeExecutionInput) if !ok { @@ -53,14 +54,31 @@ func ScriptTranscribeLambda(ctx context.Context, input any) (any, error) { for _, item := range *n { switch item.Field { case "totalDuration": - totalDuration = gconv.Int(item.Value) + if !g.IsEmpty(item.Value) { + totalDuration = gconv.Int(item.Value) + } else { + if !g.IsEmpty(item.ValueSource) { + for _, k := range item.ValueSource { + nodeConfig := nodeInput.Global.ConfigMap[k.NodeId] + if nodeConfig != nil { + for _, output := range nodeConfig.OutputResult { + if !g.IsEmpty(output[k.Field]) { + totalDuration = gconv.Int(output[k.Field]) + } + } + } + } + } + } case "modelId": modelId = gconv.Int64(item.Value) } } - // 1. 解析 valueSource 引用，填充节点输入 - ProcessValueSourceRecursive(nodeInput.Config.ModelConfig.ModelRequestParams, nodeInput.Global) + modelParams, err := BuildModelRequestBody(nodeInput.Config.ModelConfig.ModelRequestParamsPath, nodeInput.Global) + if err != nil { + return nil, err + } // 2. 构建系统提示词 + 用户输入 systemPrompt := nodeInput.Config.Prompt @@ -104,7 +122,7 @@ func ScriptTranscribeLambda(ctx context.Context, input any) (any, error) { if err != nil { return "", fmt.Errorf("获取模型配置失败: %w", err) } - result, err := gateway.ModelCallResult(ctx, nodeInput.Config.ModelConfig.ModelId, modelInfo.ModelManage.ResponseType, nodeInput.Global.SessionId, nodeInput.Config.ModelConfig.ModelRequestParams, params) + result, err := gateway.ModelCallResult(ctx, nodeInput.Config.ModelConfig.ModelId, modelInfo.ModelManage.ResponseType, nodeInput.Global.SessionId, modelParams, params) if err != nil { return nil, err } diff --git a/workflow/service/flow/lambda_value_source.go b/workflow/service/flow/lambda_value_source.go index 90df922..98d2e9d 100644 --- a/workflow/service/flow/lambda_value_source.go +++ b/workflow/service/flow/lambda_value_source.go @@ -3,9 +3,16 @@ package flow import ( "ai-agent/workflow/consts/node" flowDto "ai-agent/workflow/model/dto/flow" + "ai-agent/workflow/model/entity" + "context" + "encoding/json" + "reflect" "regexp" + "strconv" + "strings" "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/util/gconv" "github.com/tidwall/gjson" ) @@ -15,16 +22,32 @@ var ( regNumIndex = regexp.MustCompile(`\[\d+\]`) // 匹配 .attrs regAttrs = regexp.MustCompile(`\.attrs`) + // 匹配带捕获组的数组下标，转扁平点分路径用 + arrayIndexPath = regexp.MustCompile(`\[(\d+)\]`) ) // CleanFieldPath 清理字段路径：移除 .attrs、数字下标转为 [*] // 示例：usage.attrs.total_tokens → usage.total_tokens // 示例：choices.attrs[0].attrs.message.attrs.content → choices[*].message.content func CleanFieldPath(path string) string { + //// 1. 替换 [数字] 为 [*] + //s := regNumIndex.ReplaceAllString(path, `.#`) + //// 2. 移除所有 .attrs + //s = regAttrs.ReplaceAllString(s, "") + index := CleanFieldPathReplaceNumIndex(path) + attrs := CleanFieldPathRemoveAttrs(index) + return attrs +} + +func CleanFieldPathReplaceNumIndex(path string) string { // 1. 替换 [数字] 为 [*] s := regNumIndex.ReplaceAllString(path, `.#`) + return s +} + +func CleanFieldPathRemoveAttrs(path string) string { // 2. 移除所有 .attrs - s = regAttrs.ReplaceAllString(s, "") + s := regAttrs.ReplaceAllString(path, "") return s } @@ -74,7 +97,7 @@ func isSchemaEditorType(t string) bool { } // MapResultByTemplate 按 template 定义的结构，从 source 中拷贝对应字段的值。 -// 只保留 template 里出现的字段：对象字段按同名字段递归拷贝，标量/数组字段直接拷贝 source 的值。 +// 只保留 template 里出现的字段：对象字段按同名字段递归拷贝，数组字段按模板元素结构逐元素过滤，标量字段直接拷贝 source 的值。 func MapResultByTemplate(template map[string]any, source map[string]any) map[string]any { result := make(map[string]any, len(template)) for key, tmplVal := range template { @@ -88,11 +111,38 @@ func MapResultByTemplate(template map[string]any, source map[string]any) map[str } continue } + if tmplArr, isArr := tmplVal.([]any); isArr { + result[key] = mapTemplateArray(tmplArr, srcVal) + continue + } result[key] = srcVal } return result } +// mapTemplateArray 按模板数组的元素结构映射 source 数组： +// 模板首元素为对象时，逐元素按 MapResultByTemplate 过滤只保留模板字段； +// 模板数组为空或首元素非对象（无法确定元素结构）时，原样拷贝 source 数组。 +func mapTemplateArray(tmplArr []any, srcVal any) any { + srcList, ok := srcVal.([]any) + if !ok || len(tmplArr) == 0 { + return srcVal + } + elemTmpl, ok := tmplArr[0].(map[string]any) + if !ok { + return srcVal + } + result := make([]any, 0, len(srcList)) + for _, srcElem := range srcList { + if srcMap, isMap := srcElem.(map[string]any); isMap { + result = append(result, MapResultByTemplate(elemTmpl, srcMap)) + } else { + result = append(result, srcElem) + } + } + return result +} + // ProcessValueSourceRecursive 递归遍历map，同级同时存在value和valueSource则把value设置为"AA" func ProcessValueSourceRecursive(rawParams map[string]interface{}, globalParams *flowDto.FlowExecutionInput) { walkMap(rawParams, globalParams) @@ -102,9 +152,9 @@ func ProcessValueSourceRecursive(rawParams map[string]interface{}, globalParams // 返回 (value, refsName, ok)；ok=false 表示引用节点不存在或引用值仍为空。 // - 开始/表单节点：OutputConfig 平铺条目按 field == fieldName 匹配（前端约定以 field 为主， // 不兼容 path），直接读 entry 的 value / refsName -// - scriptTranscribe 节点：读 OutputResult 的 shots +// - scriptTranscribe 节点：OutputResult 是各段扁平请求参数，按段序收集字段为数组（段位留 nil） // - 其他节点：读 OutputResult 中 fieldName 路径对应的值 -func resolveValueSource(global *flowDto.FlowExecutionInput, nodeId, fieldName string) (value any, refsName any, ok bool) { +func resolveValueSource(global *flowDto.FlowExecutionInput, nodeId, field string) (value any, refsName any, ok bool) { if global == nil || global.ConfigMap == nil { return nil, nil, false } @@ -115,7 +165,7 @@ func resolveValueSource(global *flowDto.FlowExecutionInput, nodeId, fieldName st switch nodeConfig.NodeCode { case node.NodeTypeStart, node.NodeTypeForm: for _, output := range nodeConfig.OutputConfig { - if gconv.String(output["field"]) != fieldName { + if gconv.String(output["field"]) != field { continue } if !g.IsEmpty(output["value"]) { @@ -123,15 +173,32 @@ func resolveValueSource(global *flowDto.FlowExecutionInput, nodeId, fieldName st } } case node.NodeTypeScriptTranscribe: + // 脚本转写节点 OutputResult 是各段扁平请求参数（split_shots_pipeline 产出，key 为字面量 + // prompt/duration/seed 等），按段序读取 output[field] 收集为数组，供分段模型节点整体引用。 + // 每段都占一位（字段缺失/为空留 nil），保证数组与段序对齐，供 split_segment 按段取值。 + var list []any for _, output := range nodeConfig.OutputResult { - value := gjson.Get(gconv.String(output), CleanFieldPath("shots")).Value() - if !g.IsEmpty(value) { - return value, gjson.Get(gconv.String(output), CleanFieldPath("refsName")).Value(), true + list = append(list, output[field]) + } + for _, v := range list { + if !g.IsEmpty(v) { + return list, "", true } } default: for _, output := range nodeConfig.OutputResult { - value := gjson.Get(gconv.String(output), CleanFieldPath(fieldName)).Value() + // 模型节点输出记录是单 key 的字面量扁平 key（如 "choices.attrs[0].attrs.delta.attrs.content"）， + // gjson 会把 . 和 [0] 当结构路径解析，无法命中字面量 key，故先按字面量 key 直接取值； + // 未命中再回退 gjson 路径查询（兼容真正嵌套的输出结构）。 + if v, has := output[field]; has { + value = v + } else { + if field == "templates" { + value = nodeConfig.Templates + } else { + value = gjson.Get(gconv.String(output), field).Value() + } + } if !g.IsEmpty(value) { return value, gjson.Get(gconv.String(output), CleanFieldPath("refsName")).Value(), true } @@ -144,16 +211,49 @@ func resolveValueSource(global *flowDto.FlowExecutionInput, nodeId, fieldName st func walkMap(data interface{}, globalParams *flowDto.FlowExecutionInput) { switch v := data.(type) { case map[string]interface{}: - // 当前对象同时存在 value 和 valueSource + // 有 valueSource：解析引用节点值 if valueSource, hasSource := v["valueSource"]; hasSource { - mapValueSource := gconv.Map(valueSource) - nodeId := gconv.String(mapValueSource["nodeId"]) - fieldName := gconv.String(mapValueSource["fieldName"]) - if fieldName == "" { - fieldName = gconv.String(mapValueSource["field"]) - } - if nodeId != "" && fieldName != "" { - if value, refsName, ok := resolveValueSource(globalParams, nodeId, fieldName); ok { + sources := new([]entity.ValueSource) + gconv.Structs(valueSource, sources) + + // 多个引用源：把各源解析出的值拼成 "label: value"（无 label 只拼值），逗号分隔 + if len(*sources) > 1 { + parts := make([]string, 0, len(*sources)) + var refsName any + for _, src := range *sources { + value, rn, ok := resolveValueSource(globalParams, src.NodeId, src.Field) + if !ok || schemaValueEmpty(value) { + continue + } + // 引用非模型节点（开始/表单/HTTP/脚本转写等）时，值按当前字段声明的 type 做类型化转换； + // 模型节点值由模型网关处理，复制时不需要转换 + if !isModelSourceNode(globalParams, src.NodeId) { + value = assignBySchemaType(v, value) + } + text := toPlainString(value) + if src.Label != "" { + text = src.Label + ": " + text + } + parts = append(parts, text) + if !g.IsEmpty(rn) && g.IsEmpty(refsName) { + refsName = rn + } + } + if len(parts) > 0 { + v["value"] = strings.Join(parts, ", ") + if !g.IsEmpty(refsName) { + v["refsName"] = refsName + } + return + } + } else if len(*sources) == 1 { + // 单个引用源：保持旧行为，值按原样赋值（非模型节点按声明 type 转换） + src := (*sources)[0] + value, refsName, ok := resolveValueSource(globalParams, src.NodeId, src.Field) + if ok && !isModelSourceNode(globalParams, src.NodeId) { + value = assignBySchemaType(v, value) + } + if ok && !schemaValueEmpty(value) { v["value"] = value if !g.IsEmpty(refsName) { v["refsName"] = refsName @@ -162,6 +262,10 @@ func walkMap(data interface{}, globalParams *flowDto.FlowExecutionInput) { } } } + // 统一兜底：无 valueSource（或解析失败/值为空）时，value 为空或 0 则取 defaultValue + if defaultValue, hasDefault := v["defaultValue"]; hasDefault && isEmptyForFallback(v["value"]) && !schemaValueEmpty(defaultValue) { + v["value"] = defaultValue + } // 递归遍历所有子元素 for _, child := range v { walkMap(child, globalParams) @@ -174,6 +278,168 @@ func walkMap(data interface{}, globalParams *flowDto.FlowExecutionInput) { } } +// isEmptyForFallback 兜底场景判空：除 schemaValueEmpty 规则外，数字 0 也视为未填写， +// 便于配置了 defaultValue 的字段在值为 0 时用默认值兜底。 +// 覆盖 json.Number（gconv 反序列化数字的运行时类型）与字符串 "0"/"0.0"。 +func isEmptyForFallback(v interface{}) bool { + if schemaValueEmpty(v) { + return true + } + switch val := v.(type) { + case float32: + return val == 0 + case float64: + return val == 0 + case int: + return val == 0 + case int8: + return val == 0 + case int16: + return val == 0 + case int32: + return val == 0 + case int64: + return val == 0 + case uint: + return val == 0 + case uint8: + return val == 0 + case uint16: + return val == 0 + case uint32: + return val == 0 + case uint64: + return val == 0 + case json.Number: + if f, err := val.Float64(); err == nil { + return f == 0 + } + case string: + if f, err := strconv.ParseFloat(val, 64); err == nil { + return f == 0 + } + } + return false +} + +// isModelSourceNode 判断引用源节点是否为模型节点（值由模型网关处理，复制时不转换） +func isModelSourceNode(global *flowDto.FlowExecutionInput, nodeId string) bool { + if global == nil || global.ConfigMap == nil { + return false + } + nodeConfig := global.ConfigMap[nodeId] + return nodeConfig != nil && nodeConfig.NodeCode == node.NodeTypeModel +} + +// assignBySchemaType 按当前字段声明的 schema 类型把值类型化： +// string 遇数组/对象转 JSON 字符串；number/boolean 解析字符串；object/array 解析 JSON 字符串；其余原样返回 +func assignBySchemaType(node map[string]interface{}, value any) any { + t, _ := node["type"].(string) + return assignByType(t, value) +} + +// assignByType 按字段声明的 type 把值类型化；walkMap 的 schema 节点与 parseMap 的模型参数共用 +func assignByType(t string, value any) any { + switch t { + case "string": + return toSchemaString(value) + case "number": + return toSchemaNumber(value) + case "boolean": + return toSchemaBool(value) + case "object", "array": + return toSchemaStruct(value) + default: + return value + } +} + +// toSchemaString 转 string：字符串原样，数组元素拼成字符串（单元素取元素本身，多元素逗号连接），对象序列化为 JSON 字符串 +func toSchemaString(v any) any { + switch val := v.(type) { + case []interface{}: + parts := make([]string, 0, len(val)) + for _, item := range val { + parts = append(parts, toPlainString(item)) + } + return strings.Join(parts, ",") + case map[string]interface{}: + if b, err := json.Marshal(val); err == nil { + return string(b) + } + } + return v +} + +// toPlainString 把数组元素转成不带括号的纯字符串： +// 数组（[]any / 类型化切片）逐元素取纯字符串，单元素取元素本身，多元素逗号连接； +// 对象序列化为 JSON 字符串；其余原样字符串化。 +func toPlainString(v any) string { + if s, ok := v.(string); ok { + return s + } + switch val := v.(type) { + case []interface{}: + parts := make([]string, 0, len(val)) + for _, item := range val { + parts = append(parts, toPlainString(item)) + } + return strings.Join(parts, ",") + case map[string]interface{}: + if b, err := json.Marshal(val); err == nil { + return string(b) + } + } + rv := reflect.ValueOf(v) + if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) { + parts := make([]string, 0, rv.Len()) + for i := 0; i < rv.Len(); i++ { + parts = append(parts, toPlainString(rv.Index(i).Interface())) + } + return strings.Join(parts, ",") + } + if b, err := json.Marshal(v); err == nil { + return string(b) + } + return gconv.String(v) +} + +// toSchemaNumber 转 number：数字原样，字符串尝试解析为 float64，失败原样返回 +func toSchemaNumber(v any) any { + if s, ok := v.(string); ok { + if f, err := strconv.ParseFloat(s, 64); err == nil { + return f + } + } + return v +} + +// toSchemaBool 转 boolean：布尔原样，字符串尝试解析为 bool，失败原样返回 +func toSchemaBool(v any) any { + if s, ok := v.(string); ok { + if b, err := strconv.ParseBool(s); err == nil { + return b + } + } + return v +} + +// toSchemaStruct 转 object/array：合法 JSON 字符串解析为结构化数据，否则原样返回 +func toSchemaStruct(v any) any { + s, ok := v.(string) + if !ok { + return v + } + if !json.Valid([]byte(s)) { + return v + } + var out any + if err := json.Unmarshal([]byte(s), &out); err != nil { + return v + } + return out +} + // CleanEmptyModelParams 剔除模型请求参数中 value 为空的字段； // 数组/枚举（attrs / enumValues）元素整体为空时移除整个元素。0/false 视为有效值。 func CleanEmptyModelParams(params map[string]interface{}) { @@ -285,3 +551,150 @@ func schemaValueEmpty(v interface{}) bool { return false } } + +// BuildModelRequestBody 从参数定义 + 全局执行上下文构建最终嵌套 JSON 请求体。 +func BuildModelRequestBody(params []entity.FlowModelParams, globalParams *flowDto.FlowExecutionInput) (map[string]interface{}, error) { + // 1. 解析引用、过滤空值 + resolved := parseMap(params, globalParams) + + // 2. 转扁平路径映射 + flat := toFlatMap(resolved) + + // 3. 引用了脚本转写节点的字段打内部标记 __segment_fields（逗号分隔的扁平路径）， + // 供前置处理器 split_segment 按段拆批；__ 前缀内部键由 invokePreTool 统一剥离，不传给模型网关 + if seg := segmentFields(resolved, globalParams); len(seg) > 0 { + flat["__segment_fields"] = strings.Join(seg, ",") + } + + return flat, nil +} + +// segmentFields 收集引用了脚本转写节点的字段扁平路径（分段字段）。 +// 脚本转写节点按段产出一份扁平参数列表，下游模型节点单源引用其字段时， +// 值按段序聚合成数组（见 resolveValueSource 的 scriptTranscribe 分支），需随批拆分。 +func segmentFields(resolved []entity.FlowModelParams, globalParams *flowDto.FlowExecutionInput) []string { + if globalParams == nil || globalParams.ConfigMap == nil { + return nil + } + var fields []string + for _, p := range resolved { + if g.IsEmpty(p.Path) || len(p.ValueSource) != 1 { + continue + } + src := p.ValueSource[0] + if nodeConfig := globalParams.ConfigMap[src.NodeId]; nodeConfig != nil && + nodeConfig.NodeCode == node.NodeTypeScriptTranscribe { + fields = append(fields, flatPath(p.Path)) + } + } + return fields +} + +// parseMap 解析模型请求参数 +func parseMap(data []entity.FlowModelParams, globalParams *flowDto.FlowExecutionInput) []entity.FlowModelParams { + newData := make([]entity.FlowModelParams, 0, len(data)) + for _, item := range data { + var d entity.FlowModelParams + d.Path = item.Path + d.Type = item.Type + + // 无引用源：直接取静态值 + if g.IsEmpty(item.ValueSource) { + if isParamEmpty(item.Value) { + continue + } + d = item + newData = append(newData, d) + continue + } + + // 有引用源：单源保持旧行为；多源把各源解析出的值拼成 "label: value"（无 label 只拼值），逗号分隔 + var value, refsName any + if len(item.ValueSource) > 1 { + parts := make([]string, 0, len(item.ValueSource)) + for _, src := range item.ValueSource { + v, rn, ok := resolveValueSource(globalParams, src.NodeId, src.Field) + if !ok || isParamEmpty(v) { + continue + } + // 模型解析时引用其他节点的值不做类型化转换； + // 多引用源需拼接为字符串，用 toPlainString 渲染（数组取元素去括号） + text := toPlainString(v) + if src.Label != "" { + text = src.Label + ": " + text + } + parts = append(parts, text) + if !g.IsEmpty(rn) && g.IsEmpty(refsName) { + refsName = rn + } + } + if len(parts) == 0 { + // 解析失败不静默，留日志便于排查引用丢失 + glog.Debugf(context.Background(), + "resolve value source failed, nodeId=%+v path=%s", + item.ValueSource, item.Path) + continue + } + d.Value = strings.Join(parts, ", ") + d.RefsName = gconv.String(refsName) + d.ValueSource = item.ValueSource + newData = append(newData, d) + continue + } + + // 单个引用源：解析取非空值；模型解析时引用其他节点的数组值不做类型化转换，整体传给模型 + src := item.ValueSource[0] + value, refsName, ok := resolveValueSource(globalParams, src.NodeId, src.Field) + if !ok || isParamEmpty(value) { + // 解析失败不静默，留日志便于排查引用丢失 + glog.Debugf(context.Background(), + "resolve value source failed, nodeId=%+v path=%s", + item.ValueSource, item.Path) + continue + } + d.Value = value + d.RefsName = gconv.String(refsName) + d.ValueSource = item.ValueSource + newData = append(newData, d) + } + return newData +} + +// toFlatMap 将解析后的参数列表转为 sjson 可用的扁平路径映射。 +// key 为 Path，value 为参数值；保留 RefsName 供上层追踪引用来源。 +func toFlatMap(params []entity.FlowModelParams) map[string]interface{} { + m := make(map[string]interface{}, len(params)) + for _, p := range params { + if g.IsEmpty(p.Path) { + continue + } + m[flatPath(p.Path)] = p.Value + } + return m +} + +// flatPath 把数组下标路径转扁平点分路径：a[0].b → a.0.b。 +func flatPath(path string) string { + return arrayIndexPath.ReplaceAllString(path, `.$1`) +} + +// isParamEmpty 判断参数值是否为"空"。 +// 仅 nil、空字符串、空切片/映射视为空；0、false 等零值是合法值，保留。 +func isParamEmpty(v interface{}) bool { + if v == nil { + return true + } + switch val := v.(type) { + case string: + return val == "" + case []byte: + return len(val) == 0 + case []interface{}: + return len(val) == 0 + case map[string]interface{}: + return len(val) == 0 + default: + // 数字、布尔、结构体等一律视为非空 + return false + } +} diff --git a/workflow/service/flow/processor/builtin/media/media.go b/workflow/service/flow/processor/builtin/media/media.go index bc65a45..e794fb6 100644 --- a/workflow/service/flow/processor/builtin/media/media.go +++ b/workflow/service/flow/processor/builtin/media/media.go @@ -1,4 +1,4 @@ -package video +package media import ( "ai-agent/workflow/service/flow/processor" @@ -8,6 +8,7 @@ import ( "time" commonHttp "gitea.redpowerfuture.com/red-future/common/http" + "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/util/gconv" ) @@ -41,6 +42,9 @@ type mergeSubmitRes struct { TaskID string `json:"taskId"` } +// ProcessorName 处理器注册名，供模型节点前置工具分发按名判定入参形态（扁平请求体）。 +const ProcessorName = "concat_videos" + func init() { processor.Register(ConcatVideosProcessor()) } @@ -48,11 +52,12 @@ func init() { // ConcatVideosProcessor 合并视频 func ConcatVideosProcessor() *processor.Processor { return &processor.Processor{ - Name: "concat_videos", + Name: ProcessorName, Description: "合并视频", + IsShow: false, Func: func(ctx context.Context, args map[string]any) (any, error) { outputRes := parseOutputList(args) - segments, err := collectSegmentResults(outputRes) + segments, err := collectSegmentResults(ctx, outputRes) if err != nil { return nil, err } @@ -77,10 +82,16 @@ func ConcatVideosProcessor() *processor.Processor { if err != nil { return nil, err } + // 合并结果沿用输入视频的 key 返回，保持 key 不变； + // 否则下游按原 key 引用（值来源/保存文件映射）会失配 + retKey := "fileURL" + if len(outputRes) > 0 { + if k := findVideoKey(outputRes[0]); k != "" { + retKey = k + } + } return map[string]any{ - "video_url": merged.FileURL, - "duration_str": merged.DurationStr, - "task_id": merged.TaskID, + retKey: merged.FileURL, }, nil }, } @@ -217,7 +228,7 @@ type segmentResult struct { // collectSegmentResults 把模型节点/串行工具的产出（[]map[string]any）收敛为有序的分段结果列表。 // 兼容两种形状：串行工具产出的 {segment_index, video_url, duration}； // 并行模型调用产出的 {: url}（按列表顺序对应各段）。 -func collectSegmentResults(outputRes []map[string]any) ([]segmentResult, error) { +func collectSegmentResults(ctx context.Context, outputRes []map[string]any) ([]segmentResult, error) { if len(outputRes) == 0 { return nil, fmt.Errorf("没有可合并的分段视频") } @@ -226,7 +237,7 @@ func collectSegmentResults(outputRes []map[string]any) ([]segmentResult, error) seg := segmentResult{ SegmentIndex: i, Duration: gconv.Int(m["duration"]), - VideoURL: findVideoURL(m), + VideoURL: findVideoURL(ctx, m), } if idx := gconv.Int(m["segment_index"]); len(outputRes) > 1 && idx > 0 { seg.SegmentIndex = idx @@ -239,23 +250,54 @@ func collectSegmentResults(outputRes []map[string]any) ([]segmentResult, error) return segs, nil } -// findVideoURL 从模型返回参数中提取视频 URL：优先命中常见键，再兜底任意含 url 的 http 字段。 -func findVideoURL(params map[string]any) string { +// findVideoURL 从模型返回参数中提取视频 URL：优先命中常见键，再兼容扁平点号键（content.attrs.video_url 等）任意含 url 的字段。 +func findVideoURL(ctx context.Context, params map[string]any) string { + key := findVideoKey(params) + if key == "" { + return "" + } + return normalizeVideoURL(ctx, gconv.String(params[key])) +} + +// findVideoKey 返回视频 URL 所在字段的 key（命中规则与 findVideoURL 一致）， +// 供视频合并后以原 key 返回结果，避免下游按原 key 引用（值来源/保存文件映射）失配。 +func findVideoKey(params map[string]any) string { if params == nil { return "" } for _, key := range []string{"video_url", "video_oss_url", "http_file_url", "file_url", "url"} { - if v := gconv.String(params[key]); v != "" { - return v + if gconv.String(params[key]) != "" { + return key } } + // 兼容扁平点号键（如 content.attrs.video_url）：优先命中含 video 的键，再兜底任意含 url 的键 + var fallback string for k, v := range params { if !strings.Contains(strings.ToLower(k), "url") { continue } - if s := gconv.String(v); s != "" && strings.HasPrefix(s, "http") { - return s + if gconv.String(v) == "" { + continue + } + if strings.Contains(strings.ToLower(k), "video") { + return k + } + if fallback == "" { + fallback = k } } - return "" + return fallback +} + +// normalizeVideoURL 统一视频地址：已是完整 http(s) 链接原样返回；相对路径（MinIO 对象路径）补上文件前缀，供 media 服务下载 +func normalizeVideoURL(ctx context.Context, url string) string { + if url == "" || strings.HasPrefix(url, "http") { + return url + } + prefix, err := utils.GetFileAddressPrefix(ctx) + if err != nil { + g.Log().Warningf(ctx, "获取文件前缀失败，视频地址保持相对路径: %s err=%v", url, err) + return url + } + return prefix + url } diff --git a/workflow/service/flow/processor/builtin/split_batch/split_batch.go b/workflow/service/flow/processor/builtin/split_batch/split_batch.go index 466ce59..566734b 100644 --- a/workflow/service/flow/processor/builtin/split_batch/split_batch.go +++ b/workflow/service/flow/processor/builtin/split_batch/split_batch.go @@ -1,10 +1,13 @@ -// Package split_batch 工作流前置处理器：按各字段 constraint.uploadTotalMaxCount 拆分模型请求参数。 +// Package split_batch 工作流前置处理器：按默认上限（uploadTotalMaxCount 默认 15）拆分模型请求参数为多批。 +// 入参为已构建好的扁平模型请求体（BuildModelRequestBody 输出，key 为点分路径，value 已解析填充）， +// 集合字段按上限分批、元素对象取 url 为值；返回结构同入参（扁平 map 数组），未超量返回单份。 // 处理器实现自包含（算法随处理器走，不依赖业务包），通过 init 注册进 processor 注册表。 package split_batch import ( "context" "fmt" + "reflect" "sort" "github.com/gogf/gf/v2/util/gconv" @@ -12,16 +15,23 @@ import ( "ai-agent/workflow/service/flow/processor" ) +// ProcessorName 处理器注册名，供模型节点前置工具分发按名判定入参形态（扁平请求体）。 +const ProcessorName = "split_batch_model_params" + +// defaultMaxCount 每批最大元素数（constraint.uploadTotalMaxCount 当前默认值，后续动态传递）。 +const defaultMaxCount = 15 + func init() { processor.Register(SplitBatchModelParamsProcessor()) } -// SplitBatchModelParamsProcessor 将模型请求参数按各字段 constraint.uploadTotalMaxCount 分批的前置处理器。 -// 入参 args 即模型请求参数本体（需已解析好 valueSource，value 已填充）。 +// SplitBatchModelParamsProcessor 将模型请求参数按默认上限分批的前置处理器。 +// 入参 args 即扁平模型请求体（valueSource 已解析、value 已填充）。 func SplitBatchModelParamsProcessor() *processor.Processor { return &processor.Processor{ - Name: "split_batch_model_params", - Description: "按 constraint.uploadTotalMaxCount 分批模型请求参数", + Name: ProcessorName, + Description: "按最大约束构建分批模型请求数据", + IsShow: true, Func: func(ctx context.Context, args map[string]any) (any, error) { if args == nil { return nil, fmt.Errorf("缺少模型请求参数") @@ -31,37 +41,41 @@ func SplitBatchModelParamsProcessor() *processor.Processor { } } -// splitBatchField 描述一个需要按 constraint.uploadTotalMaxCount 分批的字段 -type splitBatchField struct { - path []any // 从参数根节点到该字段 value 的路径（map 用 key，数组用下标） - items []any // 该字段 value 拆出的待分批元素列表 - maxCount int // 每批最大元素数（constraint.uploadTotalMaxCount） +// batchField 描述一个需要分批的扁平字段 +type batchField struct { + key string // 扁平点分 key + items []any // 分批元素（元素为带 url 字段的对象时已取 url 为值） } -// SplitBatchModelParams 把模型请求参数拆成多批，供分批请求模型使用。 +// SplitBatchModelParams 把扁平模型请求体拆成多批，供分批请求模型使用。 // 处理流程： // 1. 深拷贝入参，避免污染调用方数据； -// 2. 递归解析 valueSource，把引用节点输出的字段值写入对应字段的 value； -// 3. 找出所有带 constraint.uploadTotalMaxCount 且 value 为集合（map/slice）的字段， -// map 按 key 排序取 value 列表作为元素；总批数 = 各字段 (元素数/上限) 向上取整的最大值； -// 4. 每批 = 整份参数深拷贝 + 各分批字段 value 替换为对应切片。 +// 2. 遍历扁平字段，值为集合（slice/array/map）的按元素数 / defaultMaxCount 向上取整， +// map 按 key 排序取 value 列表作为元素；元素为带 url 字段的对象时取 url 为值； +// 3. 总批数 = 各字段批数的最大值；每批 = 整份参数深拷贝 + 各分批字段替换为该批切片， +// 元素已耗尽的分批字段替换为空切片。 // -// 未超量时返回单份（value 已被解析填充）参数。调用方可遍历返回值逐个请求模型。 +// 未超量时返回单份参数。调用方可遍历返回值逐个请求模型。 func SplitBatchModelParams(rawParams map[string]any) []map[string]any { params, ok := deepCopyAny(rawParams).(map[string]any) if !ok { params = make(map[string]any) } - fields := make([]splitBatchField, 0) - collectBatchFields(params, nil, &fields) - + fields := make([]batchField, 0) batchCount := 1 - for _, f := range fields { - n := (len(f.items) + f.maxCount - 1) / f.maxCount + for key, v := range params { + items, has := toItems(v) + if !has { + continue + } + n := (len(items) + defaultMaxCount - 1) / defaultMaxCount if n > batchCount { batchCount = n } + if n > 1 { + fields = append(fields, batchField{key: key, items: items}) + } } if batchCount <= 1 { return []map[string]any{params} @@ -71,67 +85,46 @@ func SplitBatchModelParams(rawParams map[string]any) []map[string]any { for i := 0; i < batchCount; i++ { batch, _ := deepCopyAny(params).(map[string]any) for _, f := range fields { - start := i * f.maxCount + start := i * defaultMaxCount if start >= len(f.items) { - setValueAtPath(batch, f.path, []any{}) + batch[f.key] = []any{} continue } - end := start + f.maxCount + end := start + defaultMaxCount if end > len(f.items) { end = len(f.items) } - setValueAtPath(batch, f.path, f.items[start:end]) + batch[f.key] = f.items[start:end] } batches = append(batches, batch) } return batches } -// collectBatchFields 递归收集所有带 constraint.uploadTotalMaxCount 且 value 为集合的分批字段 -func collectBatchFields(node any, path []any, out *[]splitBatchField) { - switch v := node.(type) { - case map[string]any: - if maxCount, ok := uploadTotalMaxCountOf(v); ok { - if items, has := toItems(v["value"]); has { - *out = append(*out, splitBatchField{ - path: append(append([]any{}, path...), "value"), - items: items, - maxCount: maxCount, - }) - } - } - for key, child := range v { - collectBatchFields(child, append(append([]any{}, path...), key), out) - } - case []any: - for i, child := range v { - collectBatchFields(child, append(append([]any{}, path...), i), out) +// itemValue 取集合元素作为分批粒度时的值：元素为带 url 字段的对象时取 url（"取url为值"）， +// 其余元素原样保留。 +func itemValue(e any) any { + if m := gconv.Map(e); m != nil { + if u, ok := m["url"]; ok && u != nil { + return u } } + return e } -// uploadTotalMaxCountOf 读取 schema 节点 constraint.uploadTotalMaxCount -func uploadTotalMaxCountOf(m map[string]any) (int, bool) { - c, ok := m["constraint"] - if !ok { - return 0, false - } - cm := gconv.Map(c) - if cm == nil { - return 0, false - } - n := gconv.Int(cm["uploadTotalMaxCount"]) - if n <= 0 { - return 0, false - } - return n, true -} - -// toItems 把集合 value 转成元素列表：切片原样返回；map 按 key 排序取 value，保证分批顺序稳定 +// toItems 把集合 value 转成元素列表：切片逐元素转换（对象取 url 为值）；map 按 key 排序取 value； +// 类型化切片（[]string 等）经反射逐元素转换。 func toItems(v any) ([]any, bool) { switch val := v.(type) { case []any: - return val, len(val) > 0 + if len(val) == 0 { + return nil, false + } + items := make([]any, 0, len(val)) + for _, e := range val { + items = append(items, itemValue(e)) + } + return items, true case map[string]any: if len(val) == 0 { return nil, false @@ -143,50 +136,26 @@ func toItems(v any) ([]any, bool) { sort.Strings(keys) items := make([]any, 0, len(keys)) for _, k := range keys { - items = append(items, val[k]) + items = append(items, itemValue(val[k])) + } + return items, true + } + rv := reflect.ValueOf(v) + if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) { + n := rv.Len() + if n == 0 { + return nil, false + } + items := make([]any, 0, n) + for i := 0; i < n; i++ { + items = append(items, itemValue(rv.Index(i).Interface())) } return items, true } return nil, false } -// setValueAtPath 沿 path 逐级导航（map 用 key、数组用下标），在最后一级写入 value -func setValueAtPath(root map[string]any, path []any, value any) { - if len(path) == 0 { - return - } - var cur any = root - for i := 0; i < len(path)-1; i++ { - switch step := path[i].(type) { - case string: - m, ok := cur.(map[string]any) - if !ok { - return - } - cur = m[step] - case int: - s, ok := cur.([]any) - if !ok || step < 0 || step >= len(s) { - return - } - cur = s[step] - default: - return - } - } - switch last := path[len(path)-1].(type) { - case string: - if m, ok := cur.(map[string]any); ok { - m[last] = value - } - case int: - if s, ok := cur.([]any); ok && last >= 0 && last < len(s) { - s[last] = value - } - } -} - -// deepCopyAny 深拷贝 map[string]any / []any 嵌套结构，避免批次之间互相影响 +// deepCopyAny 深拷贝 map[string]any / []any / 类型化切片嵌套结构，避免批次之间互相影响 func deepCopyAny(v any) any { switch val := v.(type) { case map[string]any: @@ -201,7 +170,18 @@ func deepCopyAny(v any) any { res[i] = deepCopyAny(child) } return res - default: - return val } + rv := reflect.ValueOf(v) + if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) { + n := rv.Len() + out := reflect.MakeSlice(rv.Type(), n, n) + for i := 0; i < n; i++ { + d := deepCopyAny(rv.Index(i).Interface()) + if d != nil { + out.Index(i).Set(reflect.ValueOf(d)) + } + } + return out.Interface() + } + return v } diff --git a/workflow/service/flow/processor/builtin/split_segment/split_segment.go b/workflow/service/flow/processor/builtin/split_segment/split_segment.go new file mode 100644 index 0000000..b515d7e --- /dev/null +++ b/workflow/service/flow/processor/builtin/split_segment/split_segment.go @@ -0,0 +1,161 @@ +// Package split_segment 工作流前置处理器：把聚合了脚本转写各段参数的扁平请求体按段拆成多份， +// 供分段模型请求并行执行。 +// +// 脚本转写节点（split_shots_pipeline 产出）的下游模型节点单源引用其字段时， +// BuildModelRequestBody 把这些字段的数组值收集进 __segment_fields（逗号分隔的扁平路径）， +// 处理器按段序把每份参数拆成独立请求体，段与段之间互不影响（深拷贝）。 +// 处理器自包含（算法随处理器走，不依赖业务包），通过 init 注册进 processor 注册表。 +package split_segment + +import ( + "context" + "fmt" + "reflect" + "strings" + + "ai-agent/workflow/service/flow/processor" +) + +// ProcessorName 处理器注册名，供模型节点前置工具分发按名判定入参形态（扁平请求体 + __segment_fields 标记）。 +const ProcessorName = "split_segment" + +func init() { + processor.Register(SplitSegmentProcessor()) +} + +// SplitSegmentProcessor 按段拆分模型请求参数的前置处理器。 +// 入参 args 即扁平模型请求体（BuildModelRequestBody 输出，含 __segment_fields 标记）。 +func SplitSegmentProcessor() *processor.Processor { + return &processor.Processor{ + Name: ProcessorName, + Description: "按段拆分脚本转写聚合的模型请求参数", + IsShow: true, + Func: func(ctx context.Context, args map[string]any) (any, error) { + if args == nil { + return nil, fmt.Errorf("缺少模型请求参数") + } + return SplitSegmentModelParams(args) + }, + } +} + +// segField 描述一个待拆分的分段字段 +type segField struct { + path string // 扁平点分路径 + items []any // 按段序排列的元素（元素为 nil 表示该段无此字段） +} + +// SplitSegmentModelParams 把扁平模型请求体按段拆成多份，供分段请求模型使用。 +// 处理流程： +// 1. 读取 __segment_fields（逗号分隔的扁平路径），无标记则返回单份参数； +// 2. 逐个解析分段字段的数组值，要求非空且各字段长度一致（不一致直接报错，避免按错位拆分）； +// 3. 按段数深拷贝整份参数，各段覆盖其分段字段为该段元素；元素为 nil（该段无此字段）时删除该键， +// 避免把 null 传给模型网关。 +// +// 返回的每份参数仍保留 __segment_fields（由 invokePreTool 统一剥离 __ 前缀内部键）。 +func SplitSegmentModelParams(rawParams map[string]any) ([]map[string]any, error) { + paths := segmentFieldsFromArgs(rawParams) + if len(paths) == 0 { + return []map[string]any{rawParams}, nil + } + + fields := make([]segField, 0, len(paths)) + segmentCount := 0 + for _, path := range paths { + v, has := rawParams[path] + if !has { + return nil, fmt.Errorf("分段字段[%s]缺失", path) + } + items, ok := asItems(v) + if !ok || len(items) == 0 { + return nil, fmt.Errorf("分段字段[%s]值不是数组或为空", path) + } + if segmentCount == 0 { + segmentCount = len(items) + } else if len(items) != segmentCount { + return nil, fmt.Errorf("分段字段长度不一致: %s=%d, 期望 %d", path, len(items), segmentCount) + } + fields = append(fields, segField{path: path, items: items}) + } + + batches := make([]map[string]any, 0, segmentCount) + for i := 0; i < segmentCount; i++ { + batch, _ := deepCopyAny(rawParams).(map[string]any) + for _, f := range fields { + if f.items[i] == nil { + delete(batch, f.path) + continue + } + batch[f.path] = f.items[i] + } + batches = append(batches, batch) + } + return batches, nil +} + +// segmentFieldsFromArgs 解析 __segment_fields 标记为扁平路径列表，空值返回 nil。 +func segmentFieldsFromArgs(args map[string]any) []string { + raw, ok := args["__segment_fields"].(string) + if !ok || strings.TrimSpace(raw) == "" { + return nil + } + var paths []string + for _, p := range strings.Split(raw, ",") { + if t := strings.TrimSpace(p); t != "" { + paths = append(paths, t) + } + } + return paths +} + +// asItems 取分段字段值作为元素列表：[]any 直接返回（元素保持原样，nil 段位保留）； +// 类型化切片（[]string 等）经反射逐元素转 any。非切片返回 false。 +func asItems(v any) ([]any, bool) { + if list, ok := v.([]any); ok { + return list, true + } + rv := reflect.ValueOf(v) + if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) { + n := rv.Len() + if n == 0 { + return nil, true + } + items := make([]any, 0, n) + for i := 0; i < n; i++ { + items = append(items, rv.Index(i).Interface()) + } + return items, true + } + return nil, false +} + +// deepCopyAny 深拷贝 map[string]any / []any / 类型化切片嵌套结构，避免批次之间互相影响 +func deepCopyAny(v any) any { + switch val := v.(type) { + case map[string]any: + res := make(map[string]any, len(val)) + for k, child := range val { + res[k] = deepCopyAny(child) + } + return res + case []any: + res := make([]any, len(val)) + for i, child := range val { + res[i] = deepCopyAny(child) + } + return res + } + rv := reflect.ValueOf(v) + if rv.IsValid() && (rv.Kind() == reflect.Slice || rv.Kind() == reflect.Array) { + n := rv.Len() + out := reflect.MakeSlice(rv.Type(), n, n) + for i := 0; i < n; i++ { + d := deepCopyAny(rv.Index(i).Interface()) + if d != nil { + out.Index(i).Set(reflect.ValueOf(d)) + } + } + return out.Interface() + } + return v +} diff --git a/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/config.go b/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/config.go new file mode 100644 index 0000000..acdc605 --- /dev/null +++ b/workflow/service/flow/processor/builtin/split_shots_pipeline/pipeline/config.go @@ -0,0 +1,100 @@ +package pipeline + +// Config 统一浮点/阈值配置（设计 §4）。所有阈值唯一出处，阶段函数内不出现硬编码字面量。 +// 零值字段在 resolveConfig 统一由 DefaultConfig() 补齐。 +type Config struct { + // 时长计算（秒，float 允许更精细的弹性分配） + CharPerSecond float64 // 语速（字/秒），默认 4 + FastCharPerSecond float64 // 感叹/疑问多时的语速，默认 6 + SlowCharPerSecond float64 // 低落语气时的语速，默认 3 + MinDurBuffer float64 // 有声镜头时长缓冲（秒），默认 1 + MinVisualDur float64 // 纯视觉镜头最小时长（秒），默认 1 + VisualWeight float64 // 纯视觉镜头弹性权重，默认 2.0 + SpokenWeight float64 // 有声镜头弹性权重，默认 0.5 + + // 切分/断句 + SplitWindow int // 断句搜索窗口（rune），默认 60 + ShortFragmentDur float64 // 短残片阈值（秒），默认 1；时间线为整数秒，残片 ≤ 该值（通常即 1s）走并入逻辑 + BoundaryTolerance float64 // 语义切点容差（秒），<=0 时按 max(MaxSegmentDur*0.2, 2) 派生 + MaxSplitIter int // 残片并入上限重切的迭代上限，默认 10；超限返回 ErrSegmentInfeasible + + // 参考素材与截断 + MaxRefs int // 单段参考素材上限，默认 5 + MaxPromptChars int // prompt 截断长度（rune），<=0 不截断 + MinPromptFloor int // prompt 截断保底长度（rune），默认 50 + StrictInvariant bool // 时间线不变量校验：true 校验失败返回 ErrTimelineInvariant；false 仅记录告警 +} + +// DefaultConfig 返回默认配置。BoundaryTolerance 依赖 MaxSegmentDur，由 resolveConfig 派生。 +func DefaultConfig() Config { + return Config{ + CharPerSecond: 4, + FastCharPerSecond: 6, + SlowCharPerSecond: 3, + MinDurBuffer: 1, + MinVisualDur: 1, + VisualWeight: 2.0, + SpokenWeight: 0.5, + + SplitWindow: 60, + ShortFragmentDur: 1, + BoundaryTolerance: 0, // 派生子：max(MaxSegmentDur*0.2, 2) + MaxSplitIter: 10, + + MaxRefs: 5, + MaxPromptChars: 0, + MinPromptFloor: 50, + } +} + +// resolveConfig 用默认值补齐 cfg 的零值字段。maxSegmentDur 用于派生 BoundaryTolerance。 +func resolveConfig(cfg Config, maxSegmentDur int) Config { + def := DefaultConfig() + if cfg.CharPerSecond <= 0 { + cfg.CharPerSecond = def.CharPerSecond + } + if cfg.FastCharPerSecond <= 0 { + cfg.FastCharPerSecond = def.FastCharPerSecond + } + if cfg.SlowCharPerSecond <= 0 { + cfg.SlowCharPerSecond = def.SlowCharPerSecond + } + if cfg.MinDurBuffer <= 0 { + cfg.MinDurBuffer = def.MinDurBuffer + } + if cfg.MinVisualDur <= 0 { + cfg.MinVisualDur = def.MinVisualDur + } + if cfg.VisualWeight <= 0 { + cfg.VisualWeight = def.VisualWeight + } + if cfg.SpokenWeight <= 0 { + cfg.SpokenWeight = def.SpokenWeight + } + if cfg.SplitWindow <= 0 { + cfg.SplitWindow = def.SplitWindow + } + if cfg.ShortFragmentDur <= 0 { + cfg.ShortFragmentDur = def.ShortFragmentDur + } + if cfg.MaxSplitIter <= 0 { + cfg.MaxSplitIter = def.MaxSplitIter + } + if cfg.MaxRefs <= 0 { + cfg.MaxRefs = def.MaxRefs + } + if cfg.MaxPromptChars <= 0 { + cfg.MaxPromptChars = def.MaxPromptChars + } + if cfg.MinPromptFloor <= 0 { + cfg.MinPromptFloor = def.MinPromptFloor + } + if cfg.BoundaryTolerance <= 0 { + tol := float64(maxSegmentDur) * 0.2 + if tol < 2 { + tol = 2 + } + cfg.BoundaryTolerance = tol + } + return cfg +} diff --git a/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go b/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go index e369504..42ad745 100644 --- a/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go +++ b/workflow/service/flow/processor/builtin/split_shots_pipeline/split_shots_pipeline.go @@ -43,6 +43,7 @@ func SplitShotsPipelineProcessor() *processor.Processor { return &processor.Processor{ Name: "split_shots_pipeline", Description: "按 pipeline 时间线算法拆段，产出各段模型请求参数（与 split_shots 并存，灰度用）", + IsShow: false, Func: func(ctx context.Context, args map[string]any) (any, error) { input, err := parseSplitShotsInput(args) if err != nil { diff --git a/workflow/service/flow/processor/processor.go b/workflow/service/flow/processor/processor.go index 1d28d56..8945097 100644 --- a/workflow/service/flow/processor/processor.go +++ b/workflow/service/flow/processor/processor.go @@ -14,6 +14,7 @@ import ( type Processor struct { Name string Description string + IsShow bool Func func(ctx context.Context, args map[string]any) (any, error) } diff --git a/workflow/service/flow/react_ws_exec.go b/workflow/service/flow/react_ws_exec.go index df06c33..6abb864 100644 --- a/workflow/service/flow/react_ws_exec.go +++ b/workflow/service/flow/react_ws_exec.go @@ -14,6 +14,7 @@ import ( "gitea.redpowerfuture.com/red-future/common/tools" wsCommon "gitea.redpowerfuture.com/red-future/common/websocket" + "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/util/gconv" @@ -44,9 +45,41 @@ func handleToolAgent(ctx context.Context, conn *wsCommon.WsConnection, payload i return } + saveCtx := context.WithoutCancel(ctx) + + id := p.Id + if g.IsEmpty(id) { + // 会话落库：前端临时 sessionId 对应已存在会话则复用，否则新建 + err := ensureSession(saveCtx, conn.SessionId, p.Question) + if err != nil { + glog.Errorf(ctx, "会话创建失败: %v", err) + pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "会话创建失败", Error: err.Error()}) + return + } + + // 问答落库： + chatId, err := sessionDao.ExecChatDao.Insert(ctx, &sessionDto.CreateExecChatReq{ + SessionId: conn.SessionId, + RequestParams: entity.ExecChatRequestParams{Question: p.Question}, + }) + if err != nil { + glog.Errorf(ctx, "问答创建失败: %v", err) + pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "问答创建失败", Error: err.Error()}) + return + } + id = chatId + pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventRoundStart, Id: chatId}) + } + modelTools, err := tools.Default.List(ctx) if err != nil { pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "工具列表获取失败", Error: err.Error()}) + + errChat := recordChat(saveCtx, id, "", "工具列表获取失败", err, 0, 0, 0) + if errChat != nil { + glog.Errorf(ctx, "普通对话落库失败: %v", errChat) + pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "对话落库失败", Error: errChat.Error()}) + } return } systemPrompt := p.SystemPrompt @@ -62,13 +95,6 @@ func handleToolAgent(ctx context.Context, conn *wsCommon.WsConnection, payload i conn.SetMeta("toolCancel", agentCancel) defer conn.SetMeta("toolCancel", nil) defer agentCancel() - saveCtx := context.WithoutCancel(ctx) - - // 会话落库：前端临时 sessionId 对应已存在会话则复用，否则新建 - err = ensureSession(saveCtx, conn.SessionId, p.Question) - if err != nil { - pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "会话创建失败", Error: err.Error()}) - } agent := runner.NewReActAgent(p.ModelId, conn.SessionId, modelTools, systemPrompt, defaultAgentMaxStep) agent.OnEvent = func(ev runner.ReActEvent) { @@ -79,14 +105,22 @@ func handleToolAgent(ctx context.Context, conn *wsCommon.WsConnection, payload i answer, runErr := agent.Run(agentCtx, p.Question) duration := int64(time.Since(start).Seconds()) - // 前端终止：结果/错误不推前端，仅把已产生的 token 正常落库（错误记「用户已终止对话」） + // 前端终止：结果/错误不推前端，仅把已产生的 token 正常落库（友好提示记「用户已终止对话」，不记原始错误） + var errMsg string terminated := runErr != nil && errors.Is(runErr, context.Canceled) if terminated { - runErr = errChatTerminated + errMsg = errChatTerminated.Error() + runErr = nil } else if runErr != nil { + errMsg = "对话运行失败" pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "对话运行失败", Error: runErr.Error()}) } - recordChat(saveCtx, conn.SessionId, p.Question, answer, runErr, agent.TotalTokens, agent.TotalCost, duration) + err = recordChat(saveCtx, id, answer, errMsg, runErr, agent.TotalTokens, agent.TotalCost, duration) + if err != nil { + glog.Errorf(ctx, "普通对话落库失败: %v", err) + pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventError, Message: "对话落库失败", Error: err.Error()}) + } + pushAgentEvent(conn, runner.ReActEvent{Type: runner.ReActEventAnswer, Answer: answer}) } // handleToolAgentCancel 终止正在运行的对话（前端停止按钮发送 agent_cancel） @@ -106,33 +140,32 @@ func getToolCancel(conn *wsCommon.WsConnection) context.CancelFunc { // errChatTerminated 前端终止对话的错误标记（写入 exec_chat.error_message） var errChatTerminated = errors.New("用户已终止对话") -// recordChat 把一次普通对话写入 exec_chat：答案传 OSS 存 result_file_url，错误写 error_message， -// token 与费用（模型网关返回的累计 cost）落库 -func recordChat(ctx context.Context, sessionId string, question, answer string, runErr error, totalTokens int64, totalCost float64, duration int64) { - var resultFileUrl, errorMessage string +// recordChat 把一次普通对话写入 exec_chat：答案传 OSS 存 result_file_url， +// 友好提示写 error_message，原始错误写 error，token 与费用（模型网关返回的累计 cost）落库 +func recordChat(ctx context.Context, id int64, answer string, msg string, runErr error, totalTokens int64, totalCost float64, duration int64) error { + var resultFileUrl string if runErr == nil && answer != "" { - //answerJSON := gjson.New(map[string]any{"answer": answer}).MustToJson() - url, uploadErr := gateway.Upload(ctx, fmt.Sprintf("chat_%v_%d.txt", sessionId, time.Now().UnixMilli()), []byte(answer)) + url, uploadErr := gateway.Upload(ctx, fmt.Sprintf("chat_%v_%d.txt", id, time.Now().UnixMilli()), []byte(answer)) if uploadErr != nil { glog.Errorf(ctx, "普通对话答案上传OSS失败: %v", uploadErr) } else { resultFileUrl = url } - } else if runErr != nil { - errorMessage = runErr.Error() } - _, err := sessionDao.ExecChatDao.Insert(ctx, &sessionDto.CreateExecChatReq{ - SessionId: sessionId, + var errorDetail string + if runErr != nil { + errorDetail = runErr.Error() + } + _, err := sessionDao.ExecChatDao.Update(ctx, &sessionDto.UpdateExecChatReq{ + Id: id, Duration: duration, - RequestParams: entity.ExecChatRequestParams{Question: question}, ResultFileUrl: resultFileUrl, TotalTokens: int(totalTokens), TotalFee: totalCost, - ErrorMessage: errorMessage, + ErrorMessage: msg, + Error: errorDetail, }) - if err != nil { - glog.Errorf(ctx, "普通对话落库失败: %v", err) - } + return err } // pushAgentEvent 把 ReAct 过程事件转为 WS 推送消息 @@ -141,6 +174,12 @@ func pushAgentEvent(conn *wsCommon.WsConnection, ev runner.ReActEvent) { return } switch ev.Type { + case runner.ReActEventRoundStart: + _ = conn.WriteJSON(&wsCommon.WsPushMsg{ + Type: string(ev.Type), + Message: "运行开始", + Data: map[string]interface{}{"recordId": ev.Id}, + }) case runner.ReActEventModelCall: _ = conn.WriteJSON(&wsCommon.WsPushMsg{ Type: string(ev.Type), diff --git a/workflow/service/node/node_library_service.go b/workflow/service/node/node_library_service.go index 8a01936..7879844 100644 --- a/workflow/service/node/node_library_service.go +++ b/workflow/service/node/node_library_service.go @@ -8,6 +8,7 @@ import ( "ai-agent/workflow/consts/model" "ai-agent/workflow/consts/node" nodeDto "ai-agent/workflow/model/dto/node" + "ai-agent/workflow/service/flow/processor" "github.com/gogf/gf/v2/frame/g" ) @@ -23,11 +24,67 @@ func (s *nodeLibraryService) GetNodeLibrary(ctx context.Context, req *nodeDto.Wo } else { applyVideoModelOptions(tree, opts) } + // 前置/后置方法下拉填充处理器注册表数据 + if opts, err := processorOptions(ctx); err != nil { + g.Log().Warningf(ctx, "加载工作流前后置处理器选项失败，节点库降级返回: %v", err) + } else { + applyProcessorOptions(tree, opts) + } return &nodeDto.WorkflowNodeTreeRes{ Groups: tree, }, nil } +// processorOptions 从处理器注册表构建前置/后置方法下拉选项：key=处理器名，value=描述（无描述回落名称）。 +func processorOptions(ctx context.Context) ([]node.SelectOption, error) { + list, err := processor.List(ctx) + if err != nil { + return nil, err + } + opts := make([]node.SelectOption, 0, len(list)) + for _, p := range list { + if p == nil || p.Name == "" || !p.IsShow { + continue + } + value := p.Description + if value == "" { + value = p.Name + } + opts = append(opts, node.SelectOption{ + Key: p.Name, + Value: value, + }) + } + return opts, nil +} + +// applyProcessorOptions 把处理器选项填充进模型节点的 preTool/postTool 下拉。 +// 深拷贝 PreToolOption/PostToolOption 后再改，避免改写全局 NodeTypeMetaList。 +func applyProcessorOptions(tree []node.NodeGroupTree, opts []node.SelectOption) { + for gi := range tree { + for ni := range tree[gi].Nodes { + n := &tree[gi].Nodes[ni] + if n.Key != node.NodeTypeModel { + continue + } + n.PreToolOption = withProcessorOptions(n.PreToolOption, opts) + n.PostToolOption = withProcessorOptions(n.PostToolOption, opts) + return + } + } +} + +// withProcessorOptions 返回 preTool/postTool 字段列表的深拷贝，并把 select 类型字段的 Options 替换为处理器选项。 +func withProcessorOptions(fields []node.NodePresetField, opts []node.SelectOption) []node.NodePresetField { + out := append([]node.NodePresetField(nil), fields...) + for i := range out { + if out[i].Type == "select" { + out[i].Options = opts + } + } + return out +} + // videoModelOptions 按视频模型类型（600）查模型网关，转成 modelId 下拉选项：key=模型ID，value=模型名称。 func videoModelOptions(ctx context.Context) ([]node.SelectOption, error) { res, err := gateway.ListModelManage(ctx, &gateway.ListModelManageReq{ModelType: model.TypeVideo}) diff --git a/workflow/service/session/session_service.go b/workflow/service/session/session_service.go index fa96184..23c722b 100644 --- a/workflow/service/session/session_service.go +++ b/workflow/service/session/session_service.go @@ -183,6 +183,7 @@ func chatExecVO(c *entity.ExecChat) *sessionDto.VOSessionInfoResult { TotalTokens: c.TotalTokens, TotalFee: c.TotalFee, ErrorMsg: c.ErrorMessage, + Error: c.Error, CreatedAt: c.CreatedAt, } } @@ -206,6 +207,7 @@ func workflowExecVO(w *entity.ExecWorkflow, resultFileUrl string) *sessionDto.VO TotalTokens: w.TotalTokens, TotalFee: w.TotalFee, ErrorMsg: w.ErrorMessage, + Error: w.Error, CreatedAt: w.CreatedAt, } } @@ -326,3 +328,8 @@ func outputItemSuffix(ext string) string { return "内容" } } + +func (s *sessionService) ResultDelete(ctx context.Context, req *sessionDto.DeleteWorkflowResultReq) (err error) { + _, err = sessionDao.ExecWorkflowResultDao.Delete(ctx, req) + return +}