diff --git a/docs/build.md b/docs/build.md index a3cbc4a1..9ebe2779 100644 --- a/docs/build.md +++ b/docs/build.md @@ -569,7 +569,7 @@ CreateApplet（对话没有可继续的应用时） → 应用 ID CodeAssistantOffline（WAA proof） → 任务 ID StreamCodeAssistantOfflineGeneration → 步骤快照流，读到本轮回复结束或上游关闭 StreamCodeAssistantOfflineGeneration （消息列表在回复完成前结束时再请求一次） -CodeAssistantOffline、StreamCodeAssistantOfflineGeneration（回复没有正文、问题与函数调用，或上游步骤在已输出正文后失败时追加用户轮，最多 3 次） +CodeAssistantOffline、StreamCodeAssistantOfflineGeneration（回复没有正文、问题与函数调用，调用代码块缺失或无法解码，调用了被禁止的工作区工具，或上游步骤在已输出正文后失败时追加用户轮，最多 3 次） 流关闭后在后台： CancelCodeAssistantOfflineGeneration （已取得任务 ID、流没有正常解码完成且任务没有以失败步骤结束时） 巡检、删除账户、新建应用后写入记录失败： @@ -587,7 +587,7 @@ CodeAssistantOffline、StreamCodeAssistantOfflineGeneration（回复没有正文 服务按请求所用 API key 与对话历史把请求对应到应用： - 对话轮按连续的 model 轮与其余轮次分组，逐组累积 SHA-256 摘要。每组计入非思考正文 Part 直接连接后按空白归一的结果、函数调用的名称与键排序的参数、函数结果键排序的内容，其余 Part 去掉思考签名计入，内联数据以其 SHA-256 计入；调用 ID 不计入，客户端以不同格式回传同一回复时摘要相同 -- 本轮回复完整结束、没有以 `malformed_function_call` 结束、没有在函数调用之后继续输出正文、也没有在上游步骤失败后接着写时，本进程把“请求的对话 + 本轮回复的正文与函数调用”的摘要记为该应用可以继续的位置 +- 本轮回复完整结束、没有以 `malformed_function_call` 结束、没有调用被禁止的工作区内置工具、没有在函数调用之后继续输出正文、也没有在上游步骤失败后接着写时，本进程把“请求的对话 + 本轮回复的正文与函数调用”的摘要记为该应用可以继续的位置 - 新请求最后一组之前最近的 model 轮组摘要与记录相同时，在该应用中继续：field 1 只提交该组之后的轮次，field 6 为 `2`、field 14 为 `[2]`、field 21 为应用 ID；记录的位置被取走，回复完整结束后移到新回复之后，同时到达的另一请求找不到记录 - 找不到时新建应用，最后一组之前的对话轮转录在 field 9 中，见“请求映射”；包括第一轮、重新生成、编辑或压缩过的历史、回到较早的回复另起分支、另一个 API key、上一轮回复没有完整结束或在步骤失败后接着写过、最后一组过长或是 model 轮与服务重启之后；对话所在账户没有空闲槽位或不可用、请求被调度到其他账户时，放弃原应用并在新账户新建 - 新建的应用 ID 以资源类型 `assistant-applet` 记入账户运行状态。生成服务启动时与之后每 10 分钟，巡检删除闲置超过 1 小时、或本进程没有可继续对话的应用并删除记录，`DeleteApplet` 返回 404 时同样删除记录；进行中的请求所用应用保留，没有空闲槽位的账户跳过本轮；删除失败时以 WARN 记录 “闲置写代码助手应用删除失败: <错误>”，记录保留到下一轮 @@ -720,13 +720,13 @@ Part 的 field 2 为文本，field 13 为 `1` 或 `true` 时是思考。只修 3. 跳过 role 不是 `model` 的对话轮。模型轮的 Part 以（步骤序号，Part 序号）定位，同一步骤内多个模型轮的 Part 序号连续编号 4. Part 文本是该位置的完整快照：以上一次快照开头时输出新增部分，与上一次相同时不输出；不以上一次快照开头时返回协议错误 “步骤正文不是上一次快照的延续” 5. 思考 Part 的增量输出为 reasoning，其余为 text；text 来自与上一段 text 不同的步骤、且是该 Part 的第一段增量时，前面加 `\n\n` -6. 请求声明了函数时，语言标记为 `function_calls`、`function_results`、`function_call`、`tool_calls`、`tool_call` 与 `tool_code` 的代码块暂缓到以换行加 ```` ``` ```` 结束（流结束时补上结尾）再判定，围栏可以出现在行中；其他代码块按原文输出为 text，可能是围栏开头的结尾反引号暂缓输出。判定取代码块中的第一个 JSON 值，解码失败时把字符串中不合法转义（如 `\x27`，以及 `\u` 之后不是 4 位十六进制的 Windows 路径 `C:\users`）的反斜杠按字面字符、原始控制字符（如多行代码中的换行）按转义再解码一次；值为调用数组或单个调用对象时，每个元素输出为一个函数调用，调用 ID 为新的 UUID，`arguments` 为 `null` 或缺失时为 `{}`。`function_calls` 块接受任意函数名，其余标记的块只在函数均已声明时解码为调用，否则按原文输出。`function_results` 块不输出；输出函数调用之后的正文不再输出；无法解码的 `function_calls` 块不输出，回复以 `malformed_function_call` 结束 -7. 状态为 `1` 的步骤的 field 10 带 `search_web` 时，查询记入 `webSearchQueries`，每个 `[标题, 地址]` 记为一个网页来源块；同一步骤只记录一次。带 `ask_question` 且本轮还没有输出函数调用时，各问题输出为正文，选项各占一行 `- <选项>`；问题为说明中交回函数调用的那句时不输出 -8. 状态为 `1` 的步骤含模型轮时，回复记为已完成；消息中最后一个步骤状态为 `1`、含非思考正文且 field 10 为空时，本轮回复结束；输出函数调用之后模型继续输出非空白正文时，本轮回复在调用处结束，任务流关闭后取消生成任务；输出函数调用之后的步骤失败时，本轮回复同样在调用处结束，任务已经结束，不再取消。这些情况都不再读取后续消息与流尾状态 +6. 请求声明了函数时，语言标记为 `function_calls`、`function_results`、`function_call`、`tool_calls`、`tool_call` 与 `tool_code` 的代码块暂缓到以换行加 ```` ``` ```` 结束（流结束时补上结尾）再判定，围栏可以出现在行中；其他代码块按原文输出为 text，可能是围栏开头的结尾反引号暂缓输出。判定取代码块中的第一个 JSON 值，解码失败时把字符串中不合法转义（如 `\x27`，以及 `\u` 之后不是 4 位十六进制的 Windows 路径 `C:\users`）的反斜杠按字面字符、原始控制字符（如多行代码中的换行）按转义并去掉 `}` 或 `]` 前的尾随逗号再解码一次；值为调用数组或单个调用对象时，每个元素输出为一个函数调用，调用 ID 为新的 UUID，`arguments` 为 `null` 或缺失时为 `{}`。`function_calls` 块接受任意函数名，其余标记的块只在函数均已声明时解码为调用，否则按原文输出。`function_results` 块不输出；输出函数调用之后的正文不再输出；无法解码的 `function_calls` 块不输出，回复记为 `malformed_function_call` +7. 状态为 `1` 的步骤的 field 10 带 `search_web` 时，查询记入 `webSearchQueries`，每个 `[标题, 地址]` 记为一个网页来源块；同一步骤只记录一次。带 `ask_question` 且本轮还没有输出函数调用时，各问题输出为正文，选项各占一行 `- <选项>`；问题为说明中交回函数调用的那句时不输出，此时本轮还没有输出函数调用且没有待闭合的调用代码块则记为 `malformed_function_call`。请求声明了函数且本轮还没有输出函数调用时，步骤执行了 `search_web` 与 `ask_question` 以外的工作区内置工具则停止读取该任务后续步骤 +8. 状态为 `1` 的步骤含模型轮时，回复记为已完成；消息中最后一个步骤状态为 `1`、含非思考正文且 field 10 为空时，本轮回复结束；输出函数调用之后模型继续输出非空白正文、或调用了被禁止的工作区内置工具、或发出交回函数调用的 `ask_question` 而没有函数调用时，本轮任务在该步骤处结束，任务流关闭后取消生成任务；输出函数调用之后的步骤失败时，本轮回复同样在调用处结束，任务已经结束，不再取消。这些情况都不再读取后续消息与流尾状态 9. 消息列表结束后读取流尾状态，code 0 为成功，其他 code 按“错误、额度与能力边界”的表换算 HTTP 状态。上游有时在模型步骤仍为状态 `2` 时正常结束消息列表；此时以同一任务 ID 与应用再请求一次 `StreamCodeAssistantOfflineGeneration`，已完成的任务立即返回状态为 `1` 的最终快照与用量，按同一组规则继续解码，已输出的正文与函数调用不重复输出；重新打开的任务流仍在模型步骤完成前结束时，已解出函数调用或写出正文则以已有内容结束本轮，任务流关闭后取消生成任务，下一轮新建应用并带全部历史 -10. 本轮回复结束或流关闭时回复已完成：依次输出暂缓的文本、记录的搜索 grounding、用量之和，再以 `stop` 结束（无法解码的 `function_calls` 块时为 `malformed_function_call`）；回复没有完成、也没有函数调用与正文时返回协议错误 “流结束前没有完成的模型回复” +10. 本轮回复结束或流关闭时回复已完成：依次输出暂缓的文本、记录的搜索 grounding、用量之和，再以 `stop` 结束（无法解码的 `function_calls` 块或调用了被禁止的工作区内置工具且没有解出函数调用时为 `malformed_function_call`）；回复没有完成、也没有函数调用与正文时返回协议错误 “流结束前没有完成的模型回复” -一次任务的回复完成时没有输出正文、问题与函数调用（例如只有思考）且没有以 `malformed_function_call` 结束时，在同一应用中继续：field 1 只含用户轮 `Your last turn ended without any visible reply. Continue now and give your reply.`，再读取新任务的任务流，按同一组规则解码，最多 3 次；用量的输入与缓存取最后一次，输出为各次之和。 +一次任务的回复完成时没有输出正文、问题与函数调用（例如只有思考），或 `function_calls` 代码块缺失、无法解码，或调用了被禁止的工作区内置工具时，在同一应用中继续：没有正文、问题与函数调用时 field 1 为用户轮 `Your last turn ended without any visible reply. Continue now and give your reply.`；代码块缺失、无法解码或调用了被禁止的工作区内置工具时提交要求只输出 `function_calls` 代码块并调用 `ask_question` 的用户轮，已输出正文时附上 ``（最后至多 20000 个字符），后续任务开头重复已输出正文的前缀自动去掉；再读取新任务的任务流，按同一组规则解码，最多 3 次；用量的输入与缓存取最后一次，输出为各次之和。 本轮已输出正文、还没有输出函数调用时出现失败步骤且不带额度标记的，同样在同一应用中继续：field 1 只含下面的用户轮，`` 中为本轮已输出正文的最后至多 20000 个字符；新任务的输出接在已输出的正文之后，失败任务中已完成步骤的搜索来源与用量照常输出，用量按上一段的规则合并，搜索记录一并保存，与上一段的追加合计最多 3 次，仍失败时以该失败步骤的错误结束。失败之前只有思考时返回错误，由生成层重新生成： diff --git a/docs/protocol.md b/docs/protocol.md index 2da33de1..a484d798 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -772,7 +772,7 @@ ListModels 能力码 91（目录能力 `output_continuation`）的模型由官 | 18 | missing_thought_signature | 19 | malformed_response | | 22 | continuation | 其他整数 | `provider_` | -Build 通道的 Gemini API 终止原因去掉 `FINISH_REASON_` 前缀后转为上表的同名小写原因，上表没有的名称为 `provider_<小写名称>`。上游以 `malformed_function_call`、`unexpected_tool_call` 或 `malformed_response` 结束，或流在完成帧之前中断、以 5xx 流尾状态结束，且这次生成没有输出正文、函数调用与媒体时（不输出给客户端的搜索、网址上下文、地图与文件检索步骤不计入，MCP 调用计入），服务在同一账户上以同一请求重新生成，最多再生成 3 次，之后按最后一次的终止原因或中断错误结束；被重新生成取代的那一次的网页来源与引用不输出，每次重新生成从头匹配停止序列，上一次暂存的停止序列前缀不输出，重新生成的请求发送失败时本次生成以该错误结束。两个通道第一次带终止原因的帧中的内容事件标记为结束帧，自动续写中转入下一轮的帧不标记；Gemini 流式输出把结束帧的内容放进最后一块。 +Build 通道的 Gemini API 终止原因去掉 `FINISH_REASON_` 前缀后转为上表的同名小写原因，上表没有的名称为 `provider_<小写名称>`。上游以 `malformed_function_call`、`unexpected_tool_call` 或 `malformed_response` 结束，或流在完成帧之前中断、以 5xx 流尾状态结束，且这次生成没有输出正文、函数调用与媒体时（不输出给客户端的搜索、网址上下文、地图与文件检索步骤不计入，MCP 调用计入），服务在同一账户上以同一请求重新生成；已经输出纯文本且没有输出函数调用、媒体、可执行代码与服务端工具步骤时，Playground 与 Build 通道在同一账户上把已输出纯文本作为 model 轮并附上带 ``（最后至多 20000 个字符）的用户提示继续生成，续写轮开头重复已输出正文的前缀自动去掉并保留思考签名，用量按续写合并。两种情况合计最多 3 次，之后按最后一次的终止原因或中断错误结束；被重新生成取代的那一次的网页来源与引用不输出，每次重新生成从头匹配停止序列，上一次暂存的停止序列前缀不输出，重新生成的请求发送失败时本次生成以该错误结束。两个通道第一次带终止原因的帧中的内容事件标记为结束帧，自动续写中转入下一轮的帧不标记；Gemini 流式输出把结束帧的内容放进最后一块。 候选 field 5（`$[0][frame][0][0][4]`）为安全评级 `[[null, null, 类别, 概率, blocked?], ...]`，类别编号与安全设置相同，概率 `NEGLIGIBLE`=1、`LOW`=2、`MEDIUM`=3、`HIGH`=4；输入拒绝帧 `[null,[block_reason,ratings?,message]]` 的 ratings 为同形数组。Build 通道读取候选与 `promptFeedback` 的 `safetyRatings`。 diff --git a/internal/aistudio/assistant.go b/internal/aistudio/assistant.go index 617a0464..6d657ae7 100644 --- a/internal/aistudio/assistant.go +++ b/internal/aistudio/assistant.go @@ -70,9 +70,17 @@ const ( assistantLatestPrompt = "Continue the conversation from the last turn of the conversation history at the end of your instructions." // assistantEmptyReplyPrompt 是模型一轮结束时没有正文与函数调用后在同一应用中追加的用户轮 assistantEmptyReplyPrompt = "Your last turn ended without any visible reply. Continue now and give your reply." + // assistantCallFormat 是要求重写调用时附带的 function_calls 代码块格式与交回调用的说明 + assistantCallFormat = assistantCallFence + "\n[{\"name\": \"\", \"arguments\": {}}]\n```\nRight after the block, call your " + assistantQuestionTool + " tool once with the question \"" + assistantCallQuestion + "\" and write nothing more. Do not repeat any earlier text." + // assistantMalformedCallPrompt 是模型未写出 function_calls 代码块或代码块 JSON 无效时在同一应用中追加的用户轮 + assistantMalformedCallPrompt = "Your last function_calls block was missing or contained invalid JSON. Output only a valid " + assistantCallFence + " fenced JSON block:\n" + assistantCallFormat + // assistantInvalidToolPrompt 是模型调用了被禁止的远程工作区内置工具而非客户端函数时在同一应用中追加的用户轮 + assistantInvalidToolPrompt = "You called a forbidden remote workspace tool instead of a client function. Never call remote workspace tools. Output only a valid " + assistantCallFence + " fenced JSON block for the client function call(s):\n" + assistantCallFormat // assistantInterruptedPrompt 是上游步骤在已输出正文后失败时在同一应用中追加的用户轮，%s 为已输出正文的末尾 assistantInterruptedPrompt = "Your last turn was cut off by an internal error. The user already received your reply up to the end of this text:\n\n%s\n\nContinue the reply right after it without repeating it." - // assistantPartialReplyRunes 是 assistantInterruptedPrompt 带上的已输出正文末尾的字符数上限 + // assistantPartialReplyPrompt 是模型已输出部分正文后需要续出或修正工具调用时附带的已输出正文末尾，%s 为已输出正文的末尾 + assistantPartialReplyPrompt = "\n\nThe user already received your reply up to the end of this text:\n\n%s\n\nDo not repeat it." + // assistantPartialReplyRunes 是恢复提示带上的已输出正文末尾的字符数上限 assistantPartialReplyRunes = 20000 ) @@ -421,7 +429,10 @@ func (c *Client) sendAssistant(ctx context.Context, request GenerateRequest, ent var answer strings.Builder var reply []Part var usage *Usage - collect := func(event Event) error { + var finish *Event + // stripper 去掉追加任务开头重复已输出正文的前缀 + var stripper *continuationPrefixStripper + record := func(event Event) error { switch event.Kind { case EventText: answer.WriteString(event.Text) @@ -434,9 +445,16 @@ func (c *Client) sendAssistant(ctx context.Context, request GenerateRequest, ent event.Usage = &merged } usage = event.Usage + case EventFinish: + value := event + finish = &value + return nil } return emit(event) } + collect := func(event Event) error { + return stripper.filter(event, record) + } decoder := newDecoder(streamBody) err := DecodeAssistantStream(source, decoder, collect) // searches 是之前各任务完成的 search_web，interruption 是本轮接着写之前最后一次上游步骤失败 @@ -460,11 +478,26 @@ func (c *Client) sendAssistant(ctx context.Context, request GenerateRequest, ent if err != nil { break } - partial := []rune(answer.String()) - prompt = fmt.Sprintf(assistantInterruptedPrompt, string(partial[max(0, len(partial)-assistantPartialReplyRunes):])) - } else if len(reply) > replied || decoder.overran || decoder.malformed { + prompt = fmt.Sprintf(assistantInterruptedPrompt, partialReplyTail(answer.String())) + } else if !decoder.called && (decoder.malformed || decoder.invalidTool) { + if decoder.malformed { + prompt = assistantMalformedCallPrompt + } else { + prompt = assistantInvalidToolPrompt + } + if strings.TrimSpace(answer.String()) != "" { + prompt += fmt.Sprintf(assistantPartialReplyPrompt, partialReplyTail(answer.String())) + } + } else if len(reply) > replied || decoder.overran { break } + // 任务在被禁止的工具或交回调用的提问处截断时上游仍在运行，追加用户轮之前取消 + if decoder.overran && !decoder.failed { + if err = cleanup.cancel(ctx, *cleanup.generation.Load()); err != nil { + break + } + } + stripper = newContinuationPrefixStripper(answer.String()) replied = len(reply) searches = append(searches, decoder.searches...) followUp := request @@ -478,14 +511,25 @@ func (c *Client) sendAssistant(ctx context.Context, request GenerateRequest, ent err = errors.Join(DecodeAssistantStream(follow.Body, decoder, collect), follow.Body.Close()) } cleanup.completed.Store(decoder.failed || err == nil && decoder.finished && !decoder.overran) - if err == nil && interruption != nil && len(reply) == replied { + if err == nil && interruption != nil && len(reply) == replied && !decoder.malformed && !decoder.invalidTool { err = interruption } if err == nil { assistantSearches.store(apiKey, historyDigest, answer.String(), append(searches, decoder.searches...)) - if decoder.finished && !decoder.overran && !decoder.malformed && interruption == nil { + if decoder.finished && !decoder.overran && !decoder.malformed && !decoder.invalidTool && interruption == nil { cleanup.tip.Store(&assistantSessionKey{apiKey: apiKey, digest: assistantLinkDigest(links[len(links)-1].digest, true, reply)}) } + if !decoder.called && (decoder.malformed || decoder.invalidTool) { + if finish == nil { + finish = &Event{Kind: EventFinish} + } + finish.FinishReason = decodeFinishReason(10) + } + if finish != nil { + if emitErr := emit(*finish); emitErr != nil { + return emitErr + } + } } return err }, nil @@ -684,6 +728,17 @@ type assistantCleanup struct { once sync.Once } +// cancel 取消应用中的任务 generation +func (cleanup *assistantCleanup) cancel(ctx context.Context, generation string) error { + body, _ := json.Marshal([]any{generation, cleanup.session.applet}) + response, err := cleanup.client.do(ctx, "CancelCodeAssistantOfflineGeneration", cleanup.accountID, "", body, false) + if err == nil { + _, err = io.Copy(io.Discard, response.Body) + err = errors.Join(err, response.Body.Close()) + } + return err +} + // run 执行一次清理：结束会话上的请求，未完成的任务在后台以独立期限取消 func (cleanup *assistantCleanup) run() { cleanup.once.Do(func() { @@ -698,13 +753,7 @@ func (cleanup *assistantCleanup) run() { go func() { ctx, cancel := context.WithTimeout(context.Background(), assistantCleanupTimeout) defer cancel() - body, _ := json.Marshal([]any{*generation, cleanup.session.applet}) - response, err := cleanup.client.do(ctx, "CancelCodeAssistantOfflineGeneration", cleanup.accountID, "", body, false) - if err == nil { - _, err = io.Copy(io.Discard, response.Body) - err = errors.Join(err, response.Body.Close()) - } - if err != nil { + if err := cleanup.cancel(ctx, *generation); err != nil { slog.Warn("Build 写代码助手清理失败", "account", cleanup.accountID, "method", "CancelCodeAssistantOfflineGeneration", "applet", cleanup.session.applet, "error", err) } }() @@ -741,12 +790,13 @@ type AssistantStreamDecoder struct { // block 表示 held 以调用候选代码块开头，other 表示正处在其他代码块内 block bool other bool - // called 表示已输出函数调用，malformed 表示 function_calls 代码块无法解码 - called bool - malformed bool - lastStep int64 - wroteText bool - finished bool + // called 表示已输出函数调用，malformed 表示 function_calls 代码块无法解码，invalidTool 表示声明了客户端函数时模型调用了被禁止的工作区内置工具 + called bool + malformed bool + invalidTool bool + lastStep int64 + wroteText bool + finished bool // answered 表示最近解码的步骤是已完成、含正文且没有工具动作的模型步骤，即本轮回复结束 answered bool // overran 表示本轮回复在函数调用处结束：调用之后模型继续输出正文或上游后续步骤失败 @@ -856,6 +906,9 @@ func (d *AssistantStreamDecoder) Decode(raw json.RawMessage, path string) ([]Eve return nil, err } events = append(events, stepEvents...) + if d.overran { + break + } } return events, nil } @@ -897,54 +950,69 @@ func (d *AssistantStreamDecoder) decodeStep(raw json.RawMessage, path string, ev d.usage[number] = usage } var events []Event + invalidAction := false + missingCallQuestion := false if actionRaw := rawAt(step, 9); !isJSONNull(actionRaw) && status == 1 && !d.searched[number] { - if err := d.decodeSearch(actionRaw, path+"[9]", evidence); err != nil { + action, err := rawArray(actionRaw, path+"[9]", evidence) + if err != nil { + return nil, withMethod(err, assistantStreamMethod) + } + if err := d.decodeSearch(action, path+"[9]", evidence); err != nil { return nil, err } - if events, err = d.decodeQuestion(actionRaw, path+"[9]", evidence); err != nil { + var isQuestion bool + if events, isQuestion, err = d.decodeQuestion(action, path+"[9]", evidence); err != nil { return nil, err } + invalidAction = !isQuestion && isJSONNull(rawAt(action, 18)) && len(events) == 0 + missingCallQuestion = isQuestion && len(events) == 0 d.searched[number] = true } contentsRaw := rawAt(step, 0) - if isJSONNull(contentsRaw) { - return events, nil - } - contents, err := rawArray(contentsRaw, path+"[0]", evidence) - if err != nil { - return nil, withMethod(err, assistantStreamMethod) - } - partOffset := 0 - for contentIndex, contentRaw := range contents { - contentPath := fmt.Sprintf("%s[0][%d]", path, contentIndex) - content, err := rawArray(contentRaw, contentPath, evidence) + if !isJSONNull(contentsRaw) { + contents, err := rawArray(contentsRaw, path+"[0]", evidence) if err != nil { return nil, withMethod(err, assistantStreamMethod) } - role, err := rawString(rawAt(content, 1), contentPath+"[1]", evidence) - if err != nil { - return nil, withMethod(err, assistantStreamMethod) - } - if role != "model" { - continue - } - parts, err := rawArray(rawAt(content, 0), contentPath+"[0]", evidence) - if err != nil { - return nil, withMethod(err, assistantStreamMethod) - } - for partIndex, partRaw := range parts { - partEvents, err := d.decodePart(number, partOffset+partIndex, partRaw, fmt.Sprintf("%s[0][%d]", contentPath, partIndex), evidence) + partOffset := 0 + for contentIndex, contentRaw := range contents { + contentPath := fmt.Sprintf("%s[0][%d]", path, contentIndex) + content, err := rawArray(contentRaw, contentPath, evidence) if err != nil { - return nil, err + return nil, withMethod(err, assistantStreamMethod) + } + role, err := rawString(rawAt(content, 1), contentPath+"[1]", evidence) + if err != nil { + return nil, withMethod(err, assistantStreamMethod) + } + if role != "model" { + continue + } + parts, err := rawArray(rawAt(content, 0), contentPath+"[0]", evidence) + if err != nil { + return nil, withMethod(err, assistantStreamMethod) + } + for partIndex, partRaw := range parts { + partEvents, err := d.decodePart(number, partOffset+partIndex, partRaw, fmt.Sprintf("%s[0][%d]", contentPath, partIndex), evidence) + if err != nil { + return nil, err + } + events = append(events, partEvents...) + } + partOffset += len(parts) + if status == 1 { + d.finished = true } - events = append(events, partEvents...) - } - partOffset += len(parts) - if status == 1 { - d.finished = true } + d.answered = status == 1 && isJSONNull(rawAt(step, 9)) && d.wroteText && d.lastStep == number + } + if invalidAction && len(d.Functions) > 0 && !d.called { + d.invalidTool = true + d.overran = true + } else if missingCallQuestion && len(d.Functions) > 0 && !d.called && !d.block { + d.malformed = true + d.overran = true } - d.answered = status == 1 && isJSONNull(rawAt(step, 9)) && d.wroteText && d.lastStep == number return events, nil } @@ -1086,11 +1154,18 @@ func (d *AssistantStreamDecoder) text(delta string) []Event { start := strings.Index(d.held, assistantFence) if start < 0 { keep := len(d.held) - len(strings.TrimRight(d.held, "`")) - emit(d.held[:len(d.held)-keep]) + prefix := d.held[:len(d.held)-keep] + if d.malformed && strings.TrimSpace(prefix) == "" { + return events + } + emit(prefix) d.held = d.held[len(d.held)-keep:] return events } - emit(d.held[:start]) + prefix := d.held[:start] + if !d.malformed || strings.TrimSpace(prefix) != "" { + emit(prefix) + } d.held = d.held[start:] newline := strings.IndexByte(d.held, '\n') if newline < 0 { @@ -1160,7 +1235,7 @@ func assistantCalls(body string, anyName bool, functions []string) ([]Event, boo return events, true } -// repairJSONStrings 修正 JSON 字符串中模型写出的内容：不合法的转义（如按 Python 写出的 \x27、Windows 路径中 \u 之后不是十六进制的部分）改为字面反斜杠，原始控制字符（如多行代码中的换行）改为转义 +// repairJSONStrings 修正 JSON 字符串中模型写出的内容并去掉 } 或 ] 之前的尾随逗号：不合法的转义（如按 Python 写出的 \x27、Windows 路径中 \u 之后不是十六进制的部分）写为字面反斜杠，原始控制字符（如多行代码中的换行）写为转义 func repairJSONStrings(body string) string { var repaired strings.Builder quoted := false @@ -1174,6 +1249,8 @@ func repairJSONStrings(body string) string { repaired.WriteString(`\\`) case quoted && char < 0x20: fmt.Fprintf(&repaired, `\u%04x`, char) + case !quoted && char == ',' && nextNonSpaceCloses(body[index+1:]): + continue default: if char == '"' { quoted = !quoted @@ -1184,6 +1261,21 @@ func repairJSONStrings(body string) string { return repaired.String() } +// nextNonSpaceCloses 表示 rest 在跳过空白后紧跟 } 或 ] +func nextNonSpaceCloses(rest string) bool { + for index := 0; index < len(rest); index++ { + switch rest[index] { + case ' ', '\t', '\n', '\r': + continue + case '}', ']': + return true + default: + return false + } + } + return false +} + // validJSONEscape 表示反斜杠之后的 rest 以 JSON 合法转义开头，\u 之后须为 4 位十六进制 func validJSONEscape(rest string) bool { if strings.HasPrefix(rest, "u") { @@ -1193,47 +1285,43 @@ func validJSONEscape(rest string) bool { } // decodeQuestion 解码 ask_question 工具动作 field 22 [标题, null, 说明, [[参数名, 值]...]]：没有输出函数调用时把各问题与选项输出为正文，交回函数调用的提问不输出 -func (d *AssistantStreamDecoder) decodeQuestion(raw json.RawMessage, path string, evidence json.RawMessage) ([]Event, error) { - action, err := rawArray(raw, path, evidence) - if err != nil { - return nil, withMethod(err, assistantStreamMethod) - } +func (d *AssistantStreamDecoder) decodeQuestion(action []json.RawMessage, path string, evidence json.RawMessage) ([]Event, bool, error) { toolRaw := rawAt(action, 21) if isJSONNull(toolRaw) || d.called { - return nil, nil + return nil, false, nil } tool, err := rawArray(toolRaw, path+"[21]", evidence) if err != nil { - return nil, withMethod(err, assistantStreamMethod) + return nil, false, withMethod(err, assistantStreamMethod) } pairs, err := rawArray(rawAt(tool, 3), path+"[21][3]", evidence) if err != nil { - return nil, withMethod(err, assistantStreamMethod) + return nil, false, withMethod(err, assistantStreamMethod) } arguments := map[string]string{} for index, pairRaw := range pairs { pairPath := fmt.Sprintf("%s[21][3][%d]", path, index) pair, err := rawArray(pairRaw, pairPath, evidence) if err != nil { - return nil, withMethod(err, assistantStreamMethod) + return nil, false, withMethod(err, assistantStreamMethod) } name, err := rawString(rawAt(pair, 0), pairPath+"[0]", evidence) if err != nil { - return nil, withMethod(err, assistantStreamMethod) + return nil, false, withMethod(err, assistantStreamMethod) } if arguments[name], err = rawString(rawAt(pair, 1), pairPath+"[1]", evidence); err != nil { - return nil, withMethod(err, assistantStreamMethod) + return nil, false, withMethod(err, assistantStreamMethod) } } if arguments["tag"] != assistantQuestionTool { - return nil, nil + return nil, false, nil } var questions []struct { Question string `json:"question"` Options []string `json:"options"` } if err := json.Unmarshal([]byte(arguments["questions"]), &questions); err != nil { - return nil, &ProtocolEvidenceError{Method: assistantStreamMethod, Path: path + "[21][3]", Detail: "ask_question 的 questions 不是问题列表: " + err.Error(), Raw: cloneRaw(evidence)} + return nil, true, &ProtocolEvidenceError{Method: assistantStreamMethod, Path: path + "[21][3]", Detail: "ask_question 的 questions 不是问题列表: " + err.Error(), Raw: cloneRaw(evidence)} } var text strings.Builder for _, question := range questions { @@ -1249,18 +1337,14 @@ func (d *AssistantStreamDecoder) decodeQuestion(raw json.RawMessage, path string } } if text.Len() == 0 { - return nil, nil + return nil, true, nil } d.wroteText = true - return []Event{{Kind: EventText, Text: text.String()}}, nil + return []Event{{Kind: EventText, Text: text.String()}}, true, nil } // decodeSearch 把完成的 search_web 工具动作（field 19：查询、域名、摘要、[标题, 链接] 列表）并入搜索来源，查询与摘要记入 searches -func (d *AssistantStreamDecoder) decodeSearch(raw json.RawMessage, path string, evidence json.RawMessage) error { - action, err := rawArray(raw, path, evidence) - if err != nil { - return withMethod(err, assistantStreamMethod) - } +func (d *AssistantStreamDecoder) decodeSearch(action []json.RawMessage, path string, evidence json.RawMessage) error { searchRaw := rawAt(action, 18) if isJSONNull(searchRaw) { return nil @@ -1337,12 +1421,12 @@ func (d *AssistantStreamDecoder) End() ([]Event, error) { switch { case d.block: events = d.callBlock(d.held + "\n" + assistantFence) - case !d.called && d.held != "": + case !d.called && d.held != "" && (!(d.malformed || d.invalidTool) || strings.TrimSpace(d.held) != ""): events = []Event{{Kind: EventText, Text: d.held}} } events = append(events, d.recorded()...) finish := decodeFinishReason(1) - if d.malformed { + if !d.called && (d.malformed || d.invalidTool) { finish = decodeFinishReason(10) } return append(events, Event{Kind: EventFinish, FinishReason: finish}), nil diff --git a/internal/aistudio/compatibility_test.go b/internal/aistudio/compatibility_test.go index 220184ff..0b3f64dc 100644 --- a/internal/aistudio/compatibility_test.go +++ b/internal/aistudio/compatibility_test.go @@ -296,3 +296,14 @@ func TestAssistantFailureQuotaCooldown(t *testing.T) { } }) } + +// TestRepairJSONStringsTrailingComma 验证调用块 JSON 去掉 } 与 ] 之前的尾随逗号后可以解码，字符串中的逗号保持原样 +func TestRepairJSONStringsTrailingComma(t *testing.T) { + var calls []map[string]any + if err := json.Unmarshal([]byte(repairJSONStrings("[{\"name\": \"read_file\", \"arguments\": {\"path\": \"a,}b.txt\",},},\n]")), &calls); err != nil { + t.Fatal(err) + } + if len(calls) != 1 || calls[0]["name"] != "read_file" || calls[0]["arguments"].(map[string]any)["path"] != "a,}b.txt" { + t.Fatalf("calls=%v", calls) + } +} diff --git a/internal/aistudio/generate.go b/internal/aistudio/generate.go index 294d851f..0bf9eb00 100644 --- a/internal/aistudio/generate.go +++ b/internal/aistudio/generate.go @@ -867,6 +867,7 @@ func (c *Client) Generate(ctx context.Context, request GenerateRequest) (<-chan } } var usage *Usage + var previousUsage *Usage var finish *Event var output Output // turn 是按上游顺序未合并的本轮 Part 与轮次令牌，生成完整结束后保存供下一轮补回 @@ -887,9 +888,16 @@ func (c *Client) Generate(ctx context.Context, request GenerateRequest) (<-chan matchedStopSequence := "" emitEvent := func(event Event) error { turn.turnToken = cmp.Or(event.TurnToken, turn.turnToken) + if previousUsage != nil && event.FrameUsage != nil { + merged := mergeContinuedUsage(*previousUsage, *event.FrameUsage) + event.FrameUsage = &merged + } switch event.Kind { case EventUsage: value := *event.Usage + if previousUsage != nil { + value = mergeContinuedUsage(*previousUsage, value) + } usage = &value return nil case EventFinish: @@ -929,11 +937,65 @@ func (c *Client) Generate(ctx context.Context, request GenerateRequest) (<-chan return send(event) } } - emit := stopSequenceEmitter(matcher, emitEvent, &matchedStopSequence) - err := decodeStream(observeStreamActivity(ctx, response.Body), emit) - for regenerated := 0; regenerated < regenerations && regenerable(ctx, err, finish, committed); regenerated++ { + var stripper *continuationPrefixStripper + // roundUsage 是本次尝试最近一帧的上游用量，流中断时没有用量事件，续写按它合并 + var roundUsage *Usage + roundEmit := func() func(Event) error { + emit := stopSequenceEmitter(matcher, emitEvent, &matchedStopSequence) + return func(event Event) error { + turn.turnToken = cmp.Or(event.TurnToken, turn.turnToken) + roundUsage = cmp.Or(event.FrameUsage, roundUsage) + return stripper.filter(event, emit) + } + } + err := decodeStream(observeStreamActivity(ctx, response.Body), roundEmit()) + assistant := entry.model.Capabilities["assistant_route"] + for regenerated := 0; regenerated < regenerations; regenerated++ { + // 写代码助手的调用块错误已在应用内续写，这里只重新生成发送与读取失败 + if assistant && err == nil { + break + } + // emittedText 是续写时已向客户端输出的纯文本，为空时从头重新生成 + emittedText := "" + if !regenerable(ctx, err, finish, committed) { + text, ok := committedTextOnly(output) + if !ok || assistant || !regenerable(ctx, err, finish, false) { + break + } + emittedText = text + } stopClose() _ = response.Body.Close() + if emittedText != "" { + trimTrailingSignatureParts(&output.Parts) + trimTrailingSignatureParts(&turn.parts) + turn.turnToken = "" + switch { + case usage != nil: + previousUsage = usage + case roundUsage != nil && previousUsage != nil: + merged := mergeContinuedUsage(*previousUsage, *roundUsage) + previousUsage = &merged + case roundUsage != nil: + previousUsage = roundUsage + } + wireRequest = request + wireRequest.Config.StopSequences = nil + wireRequest.turnToken = "" + contents := make([]Content, 0, len(request.Contents)+2) + contents = append(contents, request.Contents...) + contents = append(contents, + Content{Role: RoleAssistant, Parts: []Part{{Text: emittedText}}}, + Content{Role: RoleUser, Parts: []Part{{Text: partialContinuationPrompt(emittedText, finish)}}}, + ) + wireRequest.Contents = contents + usage, finish = nil, nil + stripper = newContinuationPrefixStripper(emittedText) + } else { + usage, finish, sources, turn = nil, nil, nil, modelTurn{} + stripper = nil + } + roundUsage = nil if response, decodeStream, err = sendRequest(); err != nil { if ctx.Err() == nil { _ = send(Event{Kind: EventError, Err: err}) @@ -941,10 +1003,8 @@ func (c *Client) Generate(ctx context.Context, request GenerateRequest) (<-chan return } stopClose = closeOnCancel(response) - usage, finish, sources, turn = nil, nil, nil, modelTurn{} matcher = newStopSequenceMatcher(request.Config.StopSequences) - emit = stopSequenceEmitter(matcher, emitEvent, &matchedStopSequence) - err = decodeStream(observeStreamActivity(ctx, response.Body), emit) + err = decodeStream(observeStreamActivity(ctx, response.Body), roundEmit()) } if releaseSources() != nil { return @@ -971,6 +1031,9 @@ func (c *Client) Generate(ctx context.Context, request GenerateRequest) (<-chan return } if finish != nil && contentPolicyFinish(finish.FinishReason) { + if usage == nil { + usage = previousUsage + } if usage != nil && send(Event{Kind: EventUsage, Usage: usage}) != nil { return } @@ -979,6 +1042,9 @@ func (c *Client) Generate(ctx context.Context, request GenerateRequest) (<-chan } return } + if usage == nil { + usage = previousUsage + } if usage == nil { usage = localCompleteUsage(request, output) } else if usage.OutputTokensMissing { @@ -1011,6 +1077,202 @@ func contentPolicyFinish(reason string) bool { return false } +// abnormalGenerationFinish 表示结束原因属于异常工具调用或格式错误响应：格式错误的函数调用、意外的工具调用与格式错误的响应 +func abnormalGenerationFinish(reason string) bool { + switch reason { + case "malformed_function_call", "unexpected_tool_call", "malformed_response": + return true + } + return false +} + +// committedTextOnly 返回已向客户端提交的纯文本输出，没有可见文本或含函数调用、媒体、代码与服务端工具步骤时返回 false +func committedTextOnly(output Output) (string, bool) { + visible := output.Visible() + if len(visible) == 0 { + return "", false + } + var text strings.Builder + for _, part := range visible { + if part.SignatureOnly() { + continue + } + if !part.textOnly() { + return "", false + } + text.WriteString(part.Text) + } + if text.Len() == 0 { + return "", false + } + return text.String(), true +} + +// trimTrailingSignatureParts 去掉失败尝试末尾只有思考签名的 Part +func trimTrailingSignatureParts(parts *[]Part) { + items := *parts + for len(items) > 0 && items[len(items)-1].SignatureOnly() { + items = items[:len(items)-1] + } + *parts = items +} + +// partialReplyTail 返回已输出正文末尾至多 assistantPartialReplyRunes 个 Unicode 字符 +func partialReplyTail(text string) string { + runes := []rune(text) + if len(runes) > assistantPartialReplyRunes { + return string(runes[len(runes)-assistantPartialReplyRunes:]) + } + return text +} + +// partialContinuationPrompt 构造在已输出部分正文后继续生成或补发有效工具调用的用户提示 +func partialContinuationPrompt(emittedText string, finish *Event) string { + reason := "Your last turn was cut off by an internal error." + if finish != nil { + switch finish.FinishReason { + case "malformed_function_call": + reason = "Your last turn was cut off because the function call was malformed." + case "unexpected_tool_call": + reason = "Your last turn was cut off because of an unexpected tool call." + case "malformed_response": + reason = "Your last turn was cut off because the response was malformed." + } + } + return fmt.Sprintf( + "%s The user already received your reply up to the end of this text:\n\n%s\n\nContinue immediately right after without repeating it, and emit the valid tool call or remaining response.", + reason, partialReplyTail(emittedText), + ) +} + +// continuationPrefixStripper 在续写恢复轮次去掉开头重复已输出正文的前缀，尚未确定是否完全匹配时暂存前缀 +type continuationPrefixStripper struct { + targets []string + pending string + pendingSig string + active bool +} + +// newContinuationPrefixStripper 返回按已输出正文全文、去尾空白与末尾片段去重的前缀过滤器 +func newContinuationPrefixStripper(emittedText string) *continuationPrefixStripper { + if emittedText == "" { + return nil + } + tail := partialReplyTail(emittedText) + candidates := []string{ + emittedText, + strings.TrimRight(emittedText, " \t\r\n"), + tail, + strings.TrimRight(tail, " \t\r\n"), + } + targets := make([]string, 0, len(candidates)) + for _, candidate := range candidates { + if candidate == "" || slices.Contains(targets, candidate) { + continue + } + targets = append(targets, candidate) + } + if len(targets) == 0 { + return nil + } + slices.SortStableFunc(targets, func(left string, right string) int { + return cmp.Compare(len(right), len(left)) + }) + return &continuationPrefixStripper{targets: targets, active: true} +} + +// longerPrefixPending 表示 combined 是 targets[:index] 中某个更长候选的前缀 +func (s *continuationPrefixStripper) longerPrefixPending(index int, combined string) bool { + for _, longer := range s.targets[:index] { + if len(combined) < len(longer) && strings.HasPrefix(longer, combined) { + return true + } + } + return false +} + +// filter 过滤续写轮开头重复的已输出正文前缀并保留思考签名 +func (s *continuationPrefixStripper) filter(event Event, emit func(Event) error) error { + if s == nil || !s.active { + return emit(event) + } + if event.Kind != EventText { + switch event.Kind { + case EventUsage, EventReasoning, EventGrounding, EventCitation: + return emit(event) + case EventThoughtSignature: + if s.pending == "" { + return emit(event) + } + } + sig := s.pendingSig + s.pending = "" + s.pendingSig = "" + s.active = false + if sig != "" { + if err := emit(Event{Kind: EventThoughtSignature, Part: Part{ThoughtSignature: sig}}); err != nil { + return err + } + } + return emit(event) + } + if event.Transcript != nil { + s.active = false + if s.pending != "" { + if err := emit(Event{Kind: EventText, Text: s.pending, ThoughtSignature: s.pendingSig}); err != nil { + return err + } + s.pending = "" + s.pendingSig = "" + } + return emit(event) + } + combined := s.pending + event.Text + sig := cmp.Or(event.ThoughtSignature, s.pendingSig) + for index, target := range s.targets { + if strings.HasPrefix(combined, target) { + if index > 0 && s.longerPrefixPending(index, combined) { + break + } + rest := strings.TrimPrefix(combined, target) + s.pending = "" + s.pendingSig = "" + s.active = false + if rest == "" { + if sig != "" { + return emit(Event{ + Kind: EventThoughtSignature, + Part: Part{ThoughtSignature: sig}, + ProviderModel: event.ProviderModel, + TurnToken: event.TurnToken, + DebugInfo: event.DebugInfo, + ResponseID: event.ResponseID, + FrameUsage: event.FrameUsage, + FinishFrame: event.FinishFrame, + }) + } + return nil + } + event.Text = rest + event.ThoughtSignature = sig + return emit(event) + } + } + for _, target := range s.targets { + if strings.HasPrefix(target, combined) { + s.pending = combined + s.pendingSig = sig + return nil + } + } + s.pending = "" + s.pendingSig = "" + s.active = false + event.Text = combined + event.ThoughtSignature = sig + return emit(event) +} + // regenerable 表示上游以函数调用格式错误、意外的工具调用或响应格式异常结束、流在完成帧之前中断或流尾为 5xx 错误，且本次生成还没有不能丢弃的输出 func regenerable(ctx context.Context, err error, finish *Event, committed bool) bool { if err != nil { @@ -1019,7 +1281,7 @@ func regenerable(ctx context.Context, err error, finish *Event, committed bool) if ctx.Err() != nil || !errors.As(err, &interrupted) && !(errors.As(err, &rpcError) && rpcError.StatusCode >= http.StatusInternalServerError) { return false } - } else if finish == nil || finish.FinishReason != "malformed_function_call" && finish.FinishReason != "unexpected_tool_call" && finish.FinishReason != "malformed_response" { + } else if finish == nil || !abnormalGenerationFinish(finish.FinishReason) { return false } return !committed