fix(agent): 模型输出 - 校验超限并要求分批纠正
This commit is contained in:
ddcat666
@@ -125,7 +125,13 @@ func (s *Service) cloudAgentPiModel(ctx context.Context, userID, runID string, p
|
||||
if !retryable {
|
||||
return nil, err
|
||||
}
|
||||
if errors.Is(err, errCloudAgentTruncatedToolArguments) && correction == nil {
|
||||
var outputLimitErr *cloudAgentPiOutputLimitError
|
||||
if errors.As(err, &outputLimitErr) {
|
||||
correction = []map[string]any{cloudAgentRuntimeMessage(cloudAgentRuntimeContext{
|
||||
Kind: cloudAgentContextInvalidOutput, Detail: outputLimitErr.Detail,
|
||||
MaxToolCalls: cloudAgentMaxToolCalls, MaxOutputBytes: cloudAgentMaxOutputBytes,
|
||||
})}
|
||||
} else if errors.Is(err, errCloudAgentTruncatedToolArguments) {
|
||||
correction = []map[string]any{cloudAgentRuntimeMessage(cloudAgentRuntimeContext{Kind: cloudAgentContextTruncatedArguments})}
|
||||
}
|
||||
}
|
||||
@@ -168,6 +174,15 @@ var errCloudAgentAwaitingApproval = errors.New("run is waiting for approval")
|
||||
// 调用未执行，可以带纠偏上下文重做同一步。
|
||||
var errCloudAgentTruncatedToolArguments = errors.New("truncated tool arguments")
|
||||
|
||||
// cloudAgentPiOutputLimitError 标记整批未执行的输出超限，向有限重试提供具体纠正原因。
|
||||
type cloudAgentPiOutputLimitError struct {
|
||||
Detail string
|
||||
}
|
||||
|
||||
func (e *cloudAgentPiOutputLimitError) Error() string {
|
||||
return "模型输出超出限制：" + e.Detail
|
||||
}
|
||||
|
||||
// runCloudAgentModelStep 调度并等待一次模型步骤；第二个返回值表示失败是否值得重试。
|
||||
func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID string, messages []map[string]any, thinkingLevel string, correction ...map[string]any) (any, bool, error) {
|
||||
compactionSummary := cloudAgentPiCompactionSummaryRequest(messages)
|
||||
@@ -326,6 +341,16 @@ func (s *Service) runCloudAgentModelStep(ctx context.Context, userID, runID stri
|
||||
if err := json.Unmarshal([]byte(task.ResultJSON), &result); err != nil {
|
||||
return nil, false, fmt.Errorf("decode model result: %w", err)
|
||||
}
|
||||
if !compactionSummary {
|
||||
if violation := cloudAgentOutputViolation(result.Text, len(result.ToolCalls)); violation != "" {
|
||||
// 在保存 Calls 或交给运行时执行前整批拒绝；不能截取前几项造成半批写入。
|
||||
// 释放本步任务后沿用现有重试上限，模型调用仍正常计费，不执行任何工具。
|
||||
if err := s.finishCloudAgentPiModelStep(userID, runID, task.ID, "", ""); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return nil, true, &cloudAgentPiOutputLimitError{Detail: violation}
|
||||
}
|
||||
}
|
||||
finishText, finishReasoning := result.Text, result.Reasoning
|
||||
if compactionSummary {
|
||||
finishText, finishReasoning = "", ""
|
||||
|
||||
171
backend/internal/app/cloud_agent_pi_output_limit_test.go
Normal file
171
backend/internal/app/cloud_agent_pi_output_limit_test.go
Normal file
@@ -0,0 +1,171 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
"yingce/backend/internal/database"
|
||||
"yingce/backend/internal/model"
|
||||
)
|
||||
|
||||
// piOutputLimitFixture 复用现有测试库入口；PostgreSQL 只写独立临时 schema。
|
||||
// 模型结果由测试直接填入任务，不启动 Pi、HTTP 上游或真实生成。
|
||||
func piOutputLimitFixture(t *testing.T) (*Service, *gorm.DB, string) {
|
||||
t.Helper()
|
||||
var s *Service
|
||||
var db *gorm.DB
|
||||
if os.Getenv("CANVAS_TEST_POSTGRES_DSN") == "" {
|
||||
s, db, _, _ = creationTestService(t)
|
||||
} else {
|
||||
s, db = newProjectNameConflictPostgresService(t)
|
||||
s.dataDir = t.TempDir()
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||
if err := db.AutoMigrate(database.Models()...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
profile := mustEncodeModelCapabilityConfig(t, DefaultModelCapabilityConfigForModel("chat-completion", "text-test"))
|
||||
for _, row := range []any{
|
||||
&model.ModelChannel{ID: "channel", Scope: model.ChannelScopeSystem, Enabled: true},
|
||||
&model.ChannelModel{ID: "cm", ChannelID: "channel", ModelKey: "text-test", Capability: "text", Protocol: model.ChannelInterfaceChatCompletion, CapabilityConfigJSON: profile, BillingMode: "fixed_request", UnitPriceMicrocredits: 100, PriceConfigured: true, Enabled: true},
|
||||
&model.ChannelModelPriceTier{ID: "tier", ChannelModelID: "cm", SelectorKey: "{}", SelectorJSON: "{}", BillingMode: "fixed_request", UnitPriceMicrocredits: 100, PriceConfigured: true, Enabled: true},
|
||||
&model.CreditAccount{UserID: "user", AvailableMicrocredits: 10000},
|
||||
} {
|
||||
if err := db.Create(row).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
s.disablePiRuntime, s.legacyCloudAgentRootTask = true, false
|
||||
if err := os.WriteFile(filepath.Join(s.dataDir, ".settings-key"), make([]byte, 32), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Create(&model.CanvasProject{ID: "agent-canvas", UserID: "user", PayloadJSON: `{"nodes":[],"connections":[]}`}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
root, err := s.CreateCloudAgentRun("user", agentTestRequest(), "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return s, db, root.ID
|
||||
}
|
||||
|
||||
// TestPiOutputLimitCorrection 验证真实模型桥的边界、纠正上下文与有限重试。
|
||||
// 超限批次不得进入 Calls 或执行；最终失败也必须清理活动任务且给出可读原因。
|
||||
func TestPiOutputLimitCorrection(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
counts []int
|
||||
fail bool
|
||||
longText bool
|
||||
}{
|
||||
{"eight", []int{8}, false, false},
|
||||
{"nine-then-eight", []int{9, 8}, false, false},
|
||||
{"thirty-one-then-one", []int{31, 1}, false, false},
|
||||
{"bounded-failure", []int{31, 31, 31, 31}, true, false},
|
||||
{"oversized-text", []int{0, 1}, false, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
s, db, runID := piOutputLimitFixture(t)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 35*time.Second)
|
||||
defer cancel()
|
||||
type outcome struct {
|
||||
result any
|
||||
err error
|
||||
}
|
||||
done := make(chan outcome, 1)
|
||||
go func() {
|
||||
r, err := s.cloudAgentPiModel(ctx, "user", runID, map[string]json.RawMessage{"messages": json.RawMessage(`[{"role":"user","content":"继续"}]`), "thinkingLevel": json.RawMessage(`"off"`)})
|
||||
done <- outcome{r, err}
|
||||
}()
|
||||
previous := ""
|
||||
for attempt, count := range tc.counts {
|
||||
var task model.Task
|
||||
for task.ID == "" {
|
||||
_, state := agentInterjectionState(t, s, runID)
|
||||
if state.ActiveTaskID != "" && state.ActiveTaskID != previous {
|
||||
if len(state.Calls) != 0 {
|
||||
t.Fatal("rejected calls entered the checkpoint")
|
||||
}
|
||||
if err := db.First(&task, "id = ?", state.ActiveTaskID).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
break
|
||||
}
|
||||
select {
|
||||
case got := <-done:
|
||||
t.Fatalf("model bridge ended before expected task: %v", got.err)
|
||||
case <-ctx.Done():
|
||||
t.Fatal("model task not scheduled")
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
if attempt > 0 && (!strings.Contains(task.InputJSON, "invalid_output") || !strings.Contains(task.InputJSON, "maxToolCalls")) {
|
||||
t.Fatal("retry missing output-limit correction")
|
||||
}
|
||||
calls := make([]cloudAgentCall, count)
|
||||
for i := range calls {
|
||||
calls[i].ID = fmt.Sprintf("call-%d-%d", attempt, i)
|
||||
calls[i].Function.Name = "canvas_get_state"
|
||||
calls[i].Function.Arguments = "{}"
|
||||
}
|
||||
text := ""
|
||||
if tc.longText && attempt == 0 {
|
||||
text = strings.Repeat("x", cloudAgentMaxOutputBytes+1)
|
||||
}
|
||||
raw, err := json.Marshal(map[string]any{"text": text, "toolCalls": calls})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&model.Task{}).Where("id = ?", task.ID).Updates(map[string]any{"status": model.TaskStatusSucceeded, "result_json": string(raw)}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
previous = task.ID
|
||||
}
|
||||
select {
|
||||
case got := <-done:
|
||||
if tc.fail {
|
||||
if got.err == nil || !strings.Contains(got.err.Error(), "31 个工具调用") || strings.Contains(got.err.Error(), "checkpoint") {
|
||||
t.Fatalf("wrong terminal error: %v", got.err)
|
||||
}
|
||||
} else if got.err != nil {
|
||||
t.Fatal(got.err)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatal("model bridge did not finish")
|
||||
}
|
||||
_, state := agentInterjectionState(t, s, runID)
|
||||
want := tc.counts[len(tc.counts)-1]
|
||||
if tc.fail {
|
||||
want = 0
|
||||
}
|
||||
if state.ActiveTaskID != "" || len(state.Calls) != want || state.CallIndex != 0 {
|
||||
t.Fatal("output admission left invalid state")
|
||||
}
|
||||
for _, message := range state.Canonical.Messages {
|
||||
if strings.Contains(stringField(message, "content"), "invalid_output") {
|
||||
t.Fatal("retry correction leaked into canonical history")
|
||||
}
|
||||
}
|
||||
for _, event := range state.Events {
|
||||
if event.Type == "tool_completed" || event.Type == "canvas_updated" || event.Type == "approval_requested" {
|
||||
t.Fatal("model response admission executed a tool")
|
||||
}
|
||||
}
|
||||
var tasks int64
|
||||
if err := db.Model(&model.Task{}).Where("agent_run_id = ? AND operation = ?", runID, cloudAgentStepOperation).Count(&tasks).Error; err != nil || tasks != int64(len(tc.counts)) {
|
||||
t.Fatalf("unexpected retry count: %d (%v)", tasks, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,8 @@ version: 14
|
||||
|
||||
## 输出
|
||||
|
||||
每次模型回复的工具调用数量不得超过本轮执行上下文的 `maxToolCalls`（当前为 8）。任务需要更多调用时分批处理，收到这一批的真实执行结果后再继续；不要削减用户目标。该限制是工具调用数量，不是单个 `canvas_apply_ops` 内的操作数量，后者仍遵守该工具自己的参数契约。
|
||||
|
||||
默认先给结论，再说明与任务有关的画布变化和必要的下一步；纯问答无需套用操作汇报模板。用户指定输出格式时按其格式回复。没有真实写入、生成或资产就绪结果时，明确说明仍未完成。
|
||||
|
||||
## 运行状态处理
|
||||
|
||||
@@ -16,6 +16,9 @@ func TestLoadAgentPoliciesUsesDocumentMetadata(t *testing.T) {
|
||||
if strings.Contains(system.Text, "id: cloud-agent-system") || !strings.HasPrefix(system.Text, "# Cloud Agent") {
|
||||
t.Fatalf("metadata leaked into compiled policy body: %q", system.Text)
|
||||
}
|
||||
if !strings.Contains(system.Text, "maxToolCalls") || !strings.Contains(system.Text, "收到这一批的真实执行结果后再继续") {
|
||||
t.Fatal("system policy lost the tool-call batching instruction")
|
||||
}
|
||||
for _, phrase := range []string{
|
||||
"auto 只豁免其他画布修改，不豁免媒体生成",
|
||||
"所有权限模式都必须等待界面独立审批",
|
||||
|
||||
@@ -10,6 +10,11 @@ description: 信封、错误码与 OpenAPI
|
||||
自由画布和项目关联画布均使用 `canvas_projects`。Agent 运行时按登录账号和画布 ID 查询归属，不能把所属项目 ID 当作画布 ID；画布不存在、属于其他账号或数据库读取失败时，不声明读写及导出权限。
|
||||
|
||||
`canvas_updated` 的画布增量超过单事件 128 KiB 上限时，复用 `requiresRefresh: true` 通知客户端重新读取完整画布，优先保留操作摘要；摘要仍超限则仅发送刷新通知。画布正文、历史和操作回执仍完整保存，普通大小事件继续携带 `canvasPatch`。不提高事件上限、不截断正文、不修改审批暂停或模型调度判断。
|
||||
## Agent 模型输出限制
|
||||
|
||||
Pi 模型桥在保存待执行调用之前，复用单步输出限制校验：一次回复最多 8 个工具调用，正文最多 32000 字节。超限时整批未执行，释放本步模型任务并附带 `invalid_output` 及具体上限要求模型分批纠正；与既有异常重试共用首次失败后最多 3 次的上限，不截取部分调用执行。纠正请求仍是正常计费的文本模型任务，不创建媒体任务。
|
||||
|
||||
系统行为策略版本升至 14，明确工具调用数量与单个 `canvas_apply_ops` 操作数量的区别。已有会话保留其冻结策略快照；部署后用新会话验证新提示词，不修改历史检查点。
|
||||
|
||||
## 注册邮箱等价检查
|
||||
|
||||
|
||||
@@ -12,6 +12,12 @@ description: 当前项目已实现功能的手工验证清单与未确认边界
|
||||
- [x] 专项测试复现修改前大节点审批失败；修改后覆盖完整批准、执行、回执保存、正文保留、重复批准不再写入、小事件保留增量及超大摘要刷新。
|
||||
- [x] 本地用户确认原审批失败场景可以继续执行；前端画布事件与版本同步专项测试通过。
|
||||
- [ ] 验证跨浏览器、多实例及刷新失败时的恢复；本次不改既有审批恢复和模型调度流程。
|
||||
## Pi 模型输出超限
|
||||
|
||||
- 已修复：工具调用数量或正文超限时，在写入待执行调用之前整批拒绝，沿用有限重试传入具体纠正上下文，不执行前几个调用、不提高上限。
|
||||
- [x] PostgreSQL 临时 schema 在修改前复现“工具调用游标无效”；修改后验证 8 个调用允许、9/31 个调用纠正后恢复、持续超限最多 3 次重试及超长正文纠正。策略文档专项测试通过；未调用真实供应商。
|
||||
- [ ] 部署后在新会话验证大型任务按批执行；观察文本模型纠正次数及费用，失败时不声称已执行。
|
||||
- 本项独立于画布大事件修复，不改变审批和画布写入流程。
|
||||
|
||||
## 复制生图节点后节点消失
|
||||
|
||||
|
||||
Reference in New Issue
Block a user