fix(agent): 模型输出 - 提前校验超限并有限纠正

This commit is contained in:
tanzui
2026-10-09 10:27:35 +08:00
parent ac68378246
commit 8266c3b664
6 changed files with 217 additions and 3 deletions

View File

@@ -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 = "", ""

View 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)
}
})
}
}

View File

@@ -1,6 +1,6 @@
---
id: cloud-agent-system
version: 13
version: 14
---
# Cloud Agent 系统行为策略
@@ -38,6 +38,8 @@ version: 13
## 输出
每次模型回复的工具调用数量不得超过本轮执行上下文的 `maxToolCalls`(当前为 8)。任务需要更多调用时分批处理,收到这一批的真实执行结果后再继续;不要削减用户目标。该限制是工具调用数量,不是单个 `canvas_apply_ops` 内的操作数量,后者仍遵守该工具自己的参数契约。
默认先给结论,再说明与任务有关的画布变化和必要的下一步;纯问答无需套用操作汇报模板。用户指定输出格式时按其格式回复。没有真实写入、生成或资产就绪结果时,明确说明仍未完成。
## 运行状态处理

View File

@@ -10,12 +10,15 @@ func TestLoadAgentPoliciesUsesDocumentMetadata(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if system.ID != "cloud-agent-system" || system.Version != 13 || media.ID != "cloud-agent-media" || media.Version != 5 {
if system.ID != "cloud-agent-system" || system.Version != 14 || media.ID != "cloud-agent-media" || media.Version != 5 {
t.Fatalf("unexpected policy metadata: system=%+v media=%+v", system, media)
}
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 只豁免其他画布修改,不豁免媒体生成",
"所有权限模式都必须等待界面独立审批",

View File

@@ -5,6 +5,12 @@ description: 信封、错误码与 OpenAPI
# HTTP API 合同
## Agent 模型输出限制
Pi 模型桥在保存待执行调用之前,复用单步输出限制校验:一次回复最多 8 个工具调用,正文最多 32000 字节。超限时整批未执行,释放本步模型任务并附带 `invalid_output` 及具体上限要求模型分批纠正;与既有异常重试共用首次失败后最多 3 次的上限,不截取部分调用执行。纠正请求仍是正常计费的文本模型任务,不创建媒体任务。
系统行为策略版本升至 14,明确工具调用数量与单个 `canvas_apply_ops` 操作数量的区别。已有会话保留其冻结策略快照;部署后用新会话验证新提示词,不修改历史检查点。
## 注册邮箱等价检查
公开注册与邮箱绑定的唯一性检查将 `gmail.com` 和 `googlemail.com` 视为同一域,移除本地部分的点及 `+` 后缀。例如 `a.b+x@googlemail.com` 与 `ab@gmail.com` 不允许创建两个账号。其他域保留点和 `+`;现有域名白名单规则保持不变。

View File

@@ -5,6 +5,13 @@ description: 当前项目已实现功能的手工验证清单与未确认边界
# 待测试
## Pi 模型输出超限
- 已修复:工具调用数量或正文超限时,在写入待执行调用之前整批拒绝,沿用有限重试传入具体纠正上下文,不执行前几个调用、不提高上限。
- [x] PostgreSQL 临时 schema 在修改前复现“工具调用游标无效”;修改后验证 8 个调用允许、9/31 个调用纠正后恢复、持续超限最多 3 次重试及超长正文纠正。策略文档专项测试通过;未调用真实供应商。
- [ ] 部署后在新会话验证大型任务按批执行;观察文本模型纠正次数及费用,失败时不声称已执行。
- 本项独立于画布大事件修复,不改变审批和画布写入流程。
## 复制生图节点后节点消失
- 已修复:副本不再继承原节点的 `generationEffectKeys`,避免普通保存把它误判为未确认的生成结果。图片内容、提示词和参考字段仍保留,原节点不受影响。