tt/ttstd-infinite-canvas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(agent): 修复流式推理与工具参数重试

Browse Source
This commit is contained in:
ddCat
2026-09-15 05:22:07 +08:00
parent 06794aea95
commit 201f3aac65
13 changed files with 231 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/internal/app/cloud_agent_approval_preview.go
View File

@@ -54,7 +54,7 @@ type cloudAgentCanvasMutationPlan struct {
func prepareCloudAgentCanvasMutation(repo *repository.Repository, userID, canvasID string, call cloudAgentCall) (*cloudAgentCanvasMutationPlan, error) {
var args agentCanvasArgs
if err := decodeCloudAgentJSONObject(call.Function.Arguments, &args); err != nil {
return nil, BadAuthRequest("画布工具参数必须是只含支持字段的单个JSON对象")
return nil, canvasArgumentError()
}
if len(args.Ops) < 1 || len(args.Ops) > 20 || args.SnapshotHash == "" {
return nil, BadAuthRequest("画布操作数量或快照无效")

10
backend/internal/app/cloud_agent_json.go
View File

@@ -9,6 +9,16 @@ import (
var errCloudAgentJSONSingleObject = errors.New("参数必须是单个 JSON 对象")
// Only syntax/schema errors may be repaired by the model; authorization and
// unsupported mutations still fail admission before any write or approval.
type cloudAgentArgumentError struct{ error }
func (e *cloudAgentArgumentError) Unwrap() error { return e.error }
func canvasArgumentError() error {
return &cloudAgentArgumentError{BadAuthRequest("画布工具参数无效：仅允许一个 JSON 对象；顶层只含 snapshotHash 和 ops，snapshotHash 不得放入 ops。请按工具 schema 修正后重试")}
}
// decodeCloudAgentJSONObject is used for model tool arguments. Tool arguments
// are an untrusted protocol boundary: reject non-objects, unknown fields and
// trailing JSON instead of silently accepting an ambiguous payload.

9
backend/internal/app/cloud_agent_runtime.go
View File

@@ -507,6 +507,7 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error)
}
var result struct {
Text string `json:"text"`
Reasoning string `json:"reasoning"`
ToolCalls []cloudAgentCall `json:"toolCalls"`
Legacy []cloudAgentCall `json:"tool_calls"`
}
@@ -534,6 +535,9 @@ func (s *Service) advanceCloudAgent(run *model.CloudAgentExecution) (err error)
return cloudAgentSave(current, &state)
}
calls := result.ToolCalls
if result.Reasoning != "" {
state.event(run.ID, "reasoning_message", map[string]any{"messageId": task.ID + ":reasoning", "text": truncateRunes(result.Reasoning, 8000)})
}
if result.Text != "" {
state.event(run.ID, "assistant_message", map[string]any{"messageId": task.ID, "text": result.Text})
if len(calls) == 0 {
@@ -784,6 +788,11 @@ func (s *Service) advanceCloudAgentTool(run *model.CloudAgentExecution, state *c
} else {
canvasPlan, err := prepareCloudAgentCanvasMutation(repo, run.UserID, state.Request.CanvasID, call)
if err != nil {
var argumentErr *cloudAgentArgumentError
if errors.As(err, &argumentErr) {
cloudAgentToolResult(run.ID, state, call, nil, err)
return cloudAgentSave(current, state)
}
var appErr *AppError
if errors.As(err, &appErr) && appErr != nil {
return failCloudAgentAdmission(current, state, run.ID, err)

66
backend/internal/app/cloud_agent_stream.go Normal file
View File

@@ -0,0 +1,66 @@
package app
import (
"errors"
"unicode/utf8"
"gorm.io/gorm"
"infinite-canvas/backend/internal/model"
"infinite-canvas/backend/internal/repository"
)
// Publish upstream deltas to the same durable, revisioned transcript used by
// Agent SSE/replay. Each flush re-reads state; never overwrite a scheduler's
// checkpoint with the snapshot from when the model request started.
func newCloudAgentStreamPublisher(s *Service, userID, taskID, kind string) *taskTextStreamPublisher {
p := newTaskTextStreamPublisher(s, userID, taskID)
written := 0
p.sink = func(delta string) error {
remaining := 32000 - written
if remaining <= 0 {
return nil
}
if len(delta) > remaining {
delta = delta[:remaining]
for !utf8.ValidString(delta) {
delta = delta[:len(delta)-1]
}
}
if delta == "" {
return nil
}
for attempt := 0; attempt < 3; attempt++ {
run, err := s.repo.CloudAgentForActiveTask(userID, taskID)
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil
}
if err != nil {
return err
}
err = s.repo.MutateCloudAgent(userID, run.ID, run.Revision, func(current *model.CloudAgentExecution, _ *repository.Repository) error {
state, err := cloudAgentDecode(current)
if err != nil {
return err
}
if state.ActiveTaskID != taskID || (current.Status != "running" && current.Status != "queued") {
return nil
}
messageID := taskID
if kind == "reasoning_delta" {
messageID += ":reasoning"
}
state.event(run.ID, kind, map[string]any{"messageId": messageID, "text": delta})
return cloudAgentSave(current, &state)
})
if errors.Is(err, repository.ErrCreationConflict) {
continue
}
if err == nil {
written += len(delta)
}
return err
}
return repository.ErrCreationConflict
}
return p
}

5
backend/internal/app/cloud_agent_tools.go
View File

@@ -1,6 +1,7 @@
package app
import (
"crypto/sha256"
"encoding/json"
"fmt"
"sort"
@@ -103,7 +104,9 @@ func cloudAgentCanonical(system string, history []providerTextMessage, prompt st
messages = append(messages, map[string]any{"role": m.Role, "content": m.Content})
}
messages = append(messages, map[string]any{"role": "user", "content": prompt})
return canonicalAgentRequest{SystemPrompt: system, Messages: messages, Tools: cloudAgentTools(req), ToolChoice: "auto"}
// Keep routing stable across tool turns, without exposing canvas identifiers.
cacheHash := sha256.Sum256([]byte(req.CanvasID + "\x00" + system))
return canonicalAgentRequest{SystemPrompt: system, Messages: messages, Tools: cloudAgentTools(req), ToolChoice: "auto", PromptCacheKey: fmt.Sprintf("cloud-agent:%x", cacheHash[:24])}
}
func cloudAgentTools(req CloudAgentRequest) []map[string]any {
tools := []map[string]any{}

39
backend/internal/app/provider.go
View File

@@ -25,22 +25,23 @@ import (
var sseFrameBoundaryPattern = regexp.MustCompile(`\r?\n\r?\n`)
type canvasGenerationInput struct {
Mode string `json:"mode"`
Prompt string `json:"prompt"`
Config providerConfig `json:"config"`
ReferenceImages []providerMedia `json:"referenceImages"`
ReferenceVideos []providerMedia `json:"referenceVideos"`
ReferenceAudios []providerMedia `json:"referenceAudios"`
TextHistory []providerTextMessage `json:"textHistory"`
Mask *providerMedia `json:"mask"`
Metadata map[string]interface{} `json:"metadata"`
AgentRequests *agentToolRequests `json:"agentRequests"`
TextOptions canvasTextOptions `json:"textOptions"`
ImageCapability *ImageCapabilityConfig `json:"-"`
StreamText bool `json:"-"` // 分镜请求使用上游 SSE 保活；最终结构仍在流结束后统一校验。
MaxOutputTokens int `json:"-"`
OnTextDelta func(string) `json:"-"`
VideoCapability *VideoCapabilityConfig `json:"-"`
Mode string `json:"mode"`
Prompt string `json:"prompt"`
Config providerConfig `json:"config"`
ReferenceImages []providerMedia `json:"referenceImages"`
ReferenceVideos []providerMedia `json:"referenceVideos"`
ReferenceAudios []providerMedia `json:"referenceAudios"`
TextHistory []providerTextMessage `json:"textHistory"`
Mask *providerMedia `json:"mask"`
Metadata map[string]interface{} `json:"metadata"`
AgentRequests *agentToolRequests `json:"agentRequests"`
TextOptions canvasTextOptions `json:"textOptions"`
ImageCapability *ImageCapabilityConfig `json:"-"`
StreamText bool `json:"-"` // 分镜请求使用上游 SSE 保活；最终结构仍在流结束后统一校验。
MaxOutputTokens int `json:"-"`
OnTextDelta func(string) `json:"-"`
OnReasoningDelta func(string) `json:"-"`
VideoCapability *VideoCapabilityConfig `json:"-"`
}
type canvasTextOptions struct {
@@ -359,6 +360,12 @@ func (s *Service) processCanvasGenerationTask(ctx context.Context, userID string
}
if input.Mode == "text" && strings.HasPrefix(taskType, "canvas_text") && input.StreamText {
textPublisher = newTaskTextStreamPublisher(s, userID, taskExecutionID(ctx))
if input.AgentRequests != nil {
textPublisher = newCloudAgentStreamPublisher(s, userID, taskExecutionID(ctx), "assistant_delta")
reasoningPublisher := newCloudAgentStreamPublisher(s, userID, taskExecutionID(ctx), "reasoning_delta")
input.OnReasoningDelta = reasoningPublisher.Publish
defer reasoningPublisher.Close()
}
input.OnTextDelta = textPublisher.Publish
defer textPublisher.Close()
}

21
backend/internal/app/provider_agent_protocol.go
View File

@@ -11,10 +11,11 @@ import (
// The browser sends one protocol-neutral conversation. Only the selected
// provider body is materialized; declarative plugins retain their request paths.
type canonicalAgentRequest struct {
Messages []map[string]interface{} `json:"messages"`
Tools []map[string]interface{} `json:"tools"`
ToolChoice interface{} `json:"toolChoice"`
SystemPrompt string `json:"systemPrompt"`
Messages []map[string]interface{} `json:"messages"`
Tools []map[string]interface{} `json:"tools"`
ToolChoice interface{} `json:"toolChoice"`
SystemPrompt string `json:"systemPrompt"`
PromptCacheKey string `json:"promptCacheKey,omitempty"`
}
func expandCanonicalAgentRequest(source *canonicalAgentRequest, config providerConfig, declarative bool) (*agentToolRequests, error) {
@@ -155,7 +156,11 @@ func canonicalAgentChatBody(source *canonicalAgentRequest, claude bool) map[stri
if named, ok := choice.(map[string]interface{}); ok {
choice = map[string]interface{}{"type": "function", "function": map[string]interface{}{"name": named["name"]}}
}
return map[string]interface{}{"messages": messages, "tools": tools, "tool_choice": choice, "parallel_tool_calls": false}
body := map[string]interface{}{"messages": messages, "tools": tools, "tool_choice": choice, "parallel_tool_calls": false}
if !claude && source.PromptCacheKey != "" {
body["prompt_cache_key"] = source.PromptCacheKey
}
return body
}
func canonicalAgentResponsesBody(source *canonicalAgentRequest) map[string]interface{} {
@@ -183,7 +188,11 @@ func canonicalAgentResponsesBody(source *canonicalAgentRequest) map[string]inter
converted["type"] = "function"
tools = append(tools, converted)
}
return map[string]interface{}{"input": messages, "tools": tools, "tool_choice": source.ToolChoice, "parallel_tool_calls": false}
body := map[string]interface{}{"input": messages, "tools": tools, "tool_choice": source.ToolChoice, "parallel_tool_calls": false}
if source.PromptCacheKey != "" {
body["prompt_cache_key"] = source.PromptCacheKey
}
return body
}
func canonicalAgentGeminiBody(source *canonicalAgentRequest) map[string]interface{} {

8
backend/internal/app/provider_protocol.go
View File

@@ -319,6 +319,10 @@ func executeProtocolRequest(ctx context.Context, config providerConfig, spec pro
// method/path/body/auth；最终 URL 校验、凭证注入、SSRF、超时、大小限制和审计仍由宿主统一执行，
// 插件不能通过自定义请求规格绕过这些安全约束。
func executeProtocolBinaryRequest(ctx context.Context, config providerConfig, spec protocol.RequestSpec) ([]byte, string, error) {
return executeProtocolBinaryRequestWithConsumer(ctx, config, spec, nil)
}
func executeProtocolBinaryRequestWithConsumer(ctx context.Context, config providerConfig, spec protocol.RequestSpec, consume func(string, []byte)) ([]byte, string, error) {
if err := spec.Validate(); err != nil {
return nil, "", err
}
@@ -345,6 +349,10 @@ func executeProtocolBinaryRequest(ctx context.Context, config providerConfig, sp
if err := applyProtocolAuth(req, config, spec.Auth); err != nil {
return nil, "", err
}
if consume != nil {
req.Header.Set("Accept", "text/event-stream")
return doBinaryWithConsumer(req, consume)
}
return doBinary(req)
}

89
backend/internal/app/provider_text.go
View File

@@ -79,7 +79,7 @@ func runAgentToolTask(ctx context.Context, input canvasGenerationInput) (map[str
func postAgentRequest(ctx context.Context, input canvasGenerationInput, path string, body map[string]interface{}, protocol string) (map[string]interface{}, error) {
if input.StreamText {
return postStreamingAgent(ctx, input.Config, path, body, protocol, input.OnTextDelta)
return postStreamingAgent(ctx, input.Config, path, body, protocol, input.OnTextDelta, input.OnReasoningDelta)
}
delete(body, "stream")
var payload map[string]interface{}
@@ -90,7 +90,12 @@ func postAgentRequest(ctx context.Context, input canvasGenerationInput, path str
}
func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, adapter protocol.AgentAdapter) (map[string]interface{}, error) {
if input.TextOptions.Thinking {
wire := input.Config.InterfaceType
if wire == string(model.ChannelInterfaceOpenAIResponse) {
wire = "responses"
}
knownWire := wire == "chat-completion" || wire == "responses" || wire == "claude-api"
if input.TextOptions.Thinking && !knownWire {
return nil, errors.New("当前声明式 Agent 渠道尚不支持思考模式，请关闭思考模式或切换内置协议渠道")
}
if input.AgentRequests == nil {
@@ -106,6 +111,37 @@ func runDeclarativeAgentTask(ctx context.Context, input canvasGenerationInput, a
if err != nil {
return nil, err
}
if knownWire {
body := protocolBodyObject(spec.Body)
if body == nil {
return nil, errors.New("声明式 Agent 请求体必须是 JSON 对象")
}
applyTextThinking(body, input, wire)
spec.Body = body
if input.StreamText {
body["stream"] = true
if wire == "chat-completion" {
if err := ensureChatCompletionStreamUsage(body); err != nil {
return nil, err
}
}
parser := newStreamingAgentParser(wire, input.OnTextDelta)
parser.emitReasoning = input.OnReasoningDelta
data, mime, err := executeProtocolBinaryRequestWithConsumer(ctx, input.Config, spec, parser.consume)
if err != nil {
return nil, err
}
if strings.Contains(strings.ToLower(mime), "event-stream") {
parser.flush()
return parser.result()
}
var payload map[string]interface{}
if err := json.Unmarshal(data, &payload); err != nil {
return nil, fmt.Errorf("Agent 接口返回格式无效：%w", err)
}
return parseAgentToolPayload(payload, wire)
}
}
body, err := executeProtocolRequest(ctx, input.Config, spec)
if err != nil {
return nil, err
@@ -330,17 +366,17 @@ func parseAgentToolPayload(payload map[string]interface{}, protocol string) (map
return result, nil
}
func postStreamingAgent(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string)) (map[string]interface{}, error) {
func postStreamingAgent(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string), onReasoning ...func(string)) (map[string]interface{}, error) {
body["stream"] = true
if protocol == "chat-completion" {
metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
if metadata.BillingMode == "token" {
if err := ensureChatCompletionStreamUsage(body); err != nil {
return nil, err
}
if err := ensureChatCompletionStreamUsage(body); err != nil {
return nil, err
}
}
parser := newStreamingAgentParser(protocol, onDelta)
if len(onReasoning) > 0 {
parser.emitReasoning = onReasoning[0]
}
data, mimeType, err := postStreamingBinary(ctx, config, path, body, parser.consume)
if err != nil {
return nil, err
@@ -363,15 +399,16 @@ type streamingAgentToolCall struct {
}
type streamingAgentParser struct {
protocol string
buffer string
text strings.Builder
reasoning strings.Builder
toolCalls map[int]*streamingAgentToolCall
toolCallByID map[string]int
completed map[string]interface{}
err error
emit func(string)
protocol string
buffer string
text strings.Builder
reasoning strings.Builder
toolCalls map[int]*streamingAgentToolCall
toolCallByID map[string]int
completed map[string]interface{}
err error
emit func(string)
emitReasoning func(string)
}
func newStreamingAgentParser(protocol string, emit func(string)) *streamingAgentParser {
@@ -448,7 +485,7 @@ func (p *streamingAgentParser) consumeResponsesEvent(eventName string, payload m
case "response.output_text.delta", "output_text.delta":
p.appendText(stringField(payload, "delta"))
case "response.reasoning.delta", "response.reasoning_text.delta", "response.reasoning_summary_text.delta":
p.reasoning.WriteString(stringField(payload, "delta"))
p.appendReasoning(stringField(payload, "delta"))
case "response.completed":
p.completed, _ = payload["response"].(map[string]interface{})
case "response.output_item.added":
@@ -482,7 +519,7 @@ func (p *streamingAgentParser) consumeChatCompletionEvent(payload map[string]int
choice, _ := value.(map[string]interface{})
delta, _ := choice["delta"].(map[string]interface{})
p.appendText(streamContentText(delta["content"]))
p.reasoning.WriteString(firstNonEmptyString(stringField(delta, "reasoning_content"), stringField(delta, "reasoning"), stringField(delta, "reasoning_text")))
p.appendReasoning(firstNonEmptyString(stringField(delta, "reasoning_content"), stringField(delta, "reasoning"), stringField(delta, "reasoning_text")))
for fallbackIndex, toolValue := range interfaceSlice(delta["tool_calls"]) {
tool, _ := toolValue.(map[string]interface{})
index := intField(tool, "index", fallbackIndex)
@@ -509,7 +546,7 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
case "text":
p.appendText(stringField(block, "text"))
case "thinking":
p.reasoning.WriteString(firstNonEmptyString(stringField(block, "thinking"), stringField(block, "text")))
p.appendReasoning(firstNonEmptyString(stringField(block, "thinking"), stringField(block, "text")))
case "tool_use":
arguments := ""
if input := block["input"]; input != nil {
@@ -526,7 +563,7 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
p.appendText(stringField(delta, "text"))
}
if stringField(delta, "type") == "thinking_delta" {
p.reasoning.WriteString(firstNonEmptyString(stringField(delta, "thinking"), stringField(delta, "text")))
p.appendReasoning(firstNonEmptyString(stringField(delta, "thinking"), stringField(delta, "text")))
}
if stringField(delta, "type") == "input_json_delta" {
p.toolCall(index).arguments += stringField(delta, "partial_json")
@@ -551,6 +588,16 @@ func (p *streamingAgentParser) appendText(delta string) {
}
}
func (p *streamingAgentParser) appendReasoning(delta string) {
if delta == "" {
return
}
p.reasoning.WriteString(delta)
if p.emitReasoning != nil {
p.emitReasoning(delta)
}
}
func (p *streamingAgentParser) toolCall(index int) *streamingAgentToolCall {
if index < 0 {
index = 0

9
backend/internal/app/task_text_stream.go
View File

@@ -23,6 +23,7 @@ type taskTextStreamPublisher struct {
timer *time.Timer
disabled bool
closed bool
sink func(string) error
}
func newTaskTextStreamPublisher(service *Service, userID string, taskID string) *taskTextStreamPublisher {
@@ -77,7 +78,13 @@ func (p *taskTextStreamPublisher) flushLocked() {
}
content := p.buffer.String()
p.buffer.Reset()
if _, err := p.service.AppendTaskTextDelta(p.userID, p.taskID, content); err != nil {
var err error
if p.sink != nil {
err = p.sink(content)
} else {
_, err = p.service.AppendTaskTextDelta(p.userID, p.taskID, content)
}
if err != nil {
p.disabled = true
_ = p.service.log(p.userID, p.taskID, "warn", "文本流持久化已降级", err.Error())
}

6
backend/internal/repository/cloud_agent.go
View File

@@ -16,6 +16,12 @@ func (r *Repository) CloudAgent(userID, id string) (*model.CloudAgentExecution,
err := r.db.First(&run, "id = ? AND user_id = ?", id, userID).Error
return &run, err
}
func (r *Repository) CloudAgentForActiveTask(userID, taskID string) (*model.CloudAgentExecution, error) {
var run model.CloudAgentExecution
err := r.db.Where("user_id = ? AND active_task_id = ? AND status IN ?", userID, taskID, []string{"running", "queued"}).First(&run).Error
return &run, err
}
func (r *Repository) CloudAgentRoots() ([]model.Task, error) {
var tasks []model.Task
err := r.db.Where("operation = ? AND id NOT IN (SELECT id FROM cloud_agent_executions)", "cloud_agent").Order("created_at").Limit(50).Find(&tasks).Error

7
web/src/components/canvas/canvas-cloud-agent-chat-ui.tsx
View File

@@ -30,6 +30,7 @@ export type CloudAgentChatMessage = {
title?: string;
text: string;
streaming?: boolean;
reasoning?: boolean;
meta?: string;
detail?: unknown;
attachments?: CloudAgentChatAttachment[];
@@ -91,6 +92,12 @@ export function AgentChatMessage({
const isUser = item.role === "user";
const isSystem = item.role === "system";
const color = item.role === "error" ? "#ef4444" : theme.node.text;
if (item.reasoning) {
return <details className="min-w-0 text-xs" style={{ color: theme.node.muted }}>
<summary className="cursor-pointer py-1 focus-visible:outline-auto">{item.streaming ? "模型推理中" : "模型推理摘要"}</summary>
<div className="max-h-64 overflow-auto whitespace-pre-wrap break-words py-2" data-canvas-wheel-scroll>{item.text}</div>
</details>;
}
if (isSystem) {
return (
<div className="flex items-start gap-3 text-xs">

6
web/src/components/canvas/canvas-cloud-agent-panel.tsx
View File

@@ -1090,6 +1090,12 @@ function applyAgentEvent(event: AgentEvent, setMessages: Dispatch<SetStateAction
setMessages((current) => upsertTextMessage(current, String(payload.messageId || "assistant"), text, true));
return;
}
if (event.type === "reasoning_delta" || event.type === "reasoning_message") {
const id = String(payload.messageId || `${event.runId}:reasoning`);
setMessages((current) => upsertTextMessage(current, id, text, event.type === "reasoning_delta")
.map((item) => item.id === id ? { ...item, reasoning: true } : item));
return;
}
if (event.type === "assistant_message") {
setMessages((current) => upsertTextMessage(current, String(payload.messageId || event.eventId), text, false));
return;