GenarrativeAI/Genarrative
11
0
Fork 0
Code Issues 19 Pull Requests 12 Actions Packages Projects 1 Releases Wiki Activity

流式 Responses 从增量事件累积原生 output

Browse Source 
completed 未带 output 时保留 output_item 与工具参数
补空 completed 与顶层 output 的回放测试并记录排障
This commit is contained in:
孔令弘
2026-09-10 12:21:59 +00:00
parent bb010b4b1e
commit 8a93d69590
2 changed files with 184 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/project-memory/shared-memory/pitfalls.md
+8
View File
@@ -2,6 +2,14 @@
> 当前口径：本文件保留可复用的排障经验；历史条目的旧路由、旧版本和已删除文档仅作根因背景，不得据此恢复退役入口。当前命令、路由和 schema 以代码与 `docs/README.md` 为准。
## 2026-09-10 设计 Agent 需要流式 Responses 的原生 output[]，不能只靠 tool_calls
- **现象**：新策划报 `Provider 返回工具调用但未提供完整 Responses output`。模型已经在调工具，`.debug/design-agent` 里 `tool_calls` 有值但 `output` 是 `[]`。
- **原因**：设计 Agent 下一轮要把 Responses `output[]` 原样接回 `input`（`store=false` + 加密 reasoning）。旧 Planning V2 只消费 `tool_calls` 再由 Runtime 重拼 messages。官方流式常在增量里发 `output_item.added` / `function_call_arguments.done`，终态却是不带 `/response/output` 的 `response.completed`。旧解析只在 completed 里取原生数组，所以只有新策划会失败。
- **处理**：流式按 `output_index` 累积全部 output item，`arguments.done` 写回对应 item；completed 仅在带非空 `output` 时覆盖。设计 Agent 的空 output 守卫保留。
- **排查顺序**：先看 debug 响应里 `output` 是否为空、是否同时有工具调用；不要当成模型拒调工具或策划提示词错误。
- **验证**：`platform-llm` 流式夹具覆盖「空 completed + 增量 item」能拿出可回放 `responses_output`，以及 completed 顶层 `output`。
## 2026-09-05 Planning V2 审批和续跑必须等过项目锁瞬时争用
- **现象**：策划 V2 在 GDD 审批提交修改意见后提示 `项目正在被其他写操作占用：...\\.agent\\project.lock`，聊天区再出现 `项目总控 Agent 执行失败，请稍后重试`。
server-rs/crates/platform-llm/src/lib.rs
+176 -36
View File
@@ -701,6 +701,8 @@ struct ParsedStreamEvent {
// 不能借它写 finish_reason，否则会覆盖 message_delta 给出的真实 end_turn。
is_completion: bool,
tool_fragments: Vec<ToolCallFragment>,
// Responses 增量 output item。completed 若未带完整 output[]，靠这些槽位拼出可回放的原生数组。
output_items: Vec<(u64, serde_json::Value)>,
}
// 三种协议的工具调用增量归一：slot 是协议各自的索引（Chat/Anthropic 的 index、
@@ -1908,7 +1910,7 @@ impl LlmClient {
response_id,
usage: accumulation.usage,
tool_calls,
responses_output: accumulation.responses_output,
responses_output: compact_responses_output(accumulation.responses_output),
})
}
@@ -2256,11 +2258,15 @@ where
is_terminal,
is_completion,
tool_fragments,
output_items,
} = event;
if let Some(output) = responses_output {
accumulation.responses_output = output;
}
for (slot, item) in output_items {
upsert_responses_output_item(&mut accumulation.responses_output, slot, item);
}
if is_completion {
accumulation.completion_observed = true;
@@ -2289,6 +2295,13 @@ where
// 工具调用只累加，不进 on_delta：调用方的流式通道仍然只承载文本。
let ids_repeated_in_event = tool_fragment_ids_repeated_in_event(&tool_fragments);
for fragment in tool_fragments {
if let Some(arguments) = fragment.arguments_complete.as_deref() {
patch_responses_output_arguments(
&mut accumulation.responses_output,
fragment.slot,
arguments,
);
}
accumulation.push_tool_fragment(fragment, &ids_repeated_in_event)?;
}
@@ -3632,41 +3645,16 @@ fn parse_responses_sse_event(data: &str) -> Result<Option<ParsedStreamEvent>, Ll
// 工具会让纯文本的 completed-only 响应变成 EmptyResponse，让「正文 + 工具」
// 响应静默丢掉模型的前置说明。
text_snapshot: extract_responses_terminal_text(&parsed),
responses_output: parsed
.pointer("/response/output")
.and_then(serde_json::Value::as_array)
.cloned(),
responses_output: extract_responses_output_array(&parsed),
tool_fragments: extract_responses_completed_tool_fragments(&parsed)?,
..Default::default()
})),
// 工具调用先由 output_item.added 宣告身份，再用 arguments delta 拼参数；
// .done 给出权威完整参数，用它覆盖拼接结果。三个事件共用 output_index 作为槽位。
"response.output_item.added" => {
let item = parsed.get("item");
if item
.and_then(|item| item.get("type"))
.and_then(serde_json::Value::as_str)
!= Some("function_call")
{
return Ok(None);
}
let slot = responses_output_slot(&parsed)
.ok_or_else(|| missing_tool_slot_error("Responses", event_type, "output_index"))?;
Ok(Some(ParsedStreamEvent {
tool_fragments: vec![ToolCallFragment {
slot,
id: item
.and_then(|item| item.get("call_id").or_else(|| item.get("id")))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
name: item
.and_then(|item| item.get("name"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
..Default::default()
}],
..Default::default()
}))
// output_item.added / done 同时把原生 item 按槽位写入，供 completed 未带
// output[] 时回放；function_call 仍走 tool_fragments 归并。
"response.output_item.added" | "response.output_item.done" => {
parse_responses_output_item_event(&parsed, event_type)
}
"response.function_call_arguments.delta" => {
let slot = responses_output_slot(&parsed)
@@ -3732,14 +3720,82 @@ fn extract_responses_terminal_text(parsed: &serde_json::Value) -> Option<String>
extract_responses_text(&envelope).filter(|text| !text.trim().is_empty())
}
fn extract_responses_output_array(parsed: &serde_json::Value) -> Option<Vec<serde_json::Value>> {
parsed
.pointer("/response/output")
.or_else(|| parsed.get("output"))
.and_then(serde_json::Value::as_array)
.filter(|items| !items.is_empty())
.cloned()
}
fn upsert_responses_output_item(
output: &mut Vec<serde_json::Value>,
slot: u64,
item: serde_json::Value,
) {
let index = slot as usize;
if index >= output.len() {
output.resize(index + 1, serde_json::Value::Null);
}
output[index] = item;
}
fn patch_responses_output_arguments(output: &mut [serde_json::Value], slot: u64, arguments: &str) {
let Some(serde_json::Value::Object(map)) = output.get_mut(slot as usize) else {
return;
};
map.insert(
"arguments".to_string(),
serde_json::Value::String(arguments.to_string()),
);
}
fn compact_responses_output(mut output: Vec<serde_json::Value>) -> Vec<serde_json::Value> {
output.retain(|item| !item.is_null());
output
}
fn parse_responses_output_item_event(
parsed: &serde_json::Value,
event_type: &str,
) -> Result<Option<ParsedStreamEvent>, LlmError> {
let Some(item) = parsed.get("item").filter(|item| item.is_object()).cloned() else {
return Ok(None);
};
let slot = responses_output_slot(parsed)
.ok_or_else(|| missing_tool_slot_error("Responses", event_type, "output_index"))?;
let mut event = ParsedStreamEvent {
output_items: vec![(slot, item.clone())],
..Default::default()
};
if item.get("type").and_then(serde_json::Value::as_str) == Some("function_call") {
event.tool_fragments = vec![ToolCallFragment {
slot,
id: item
.get("call_id")
.or_else(|| item.get("id"))
.and_then(serde_json::Value::as_str)
.map(str::to_string),
name: item
.get("name")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
arguments_complete: item
.get("arguments")
.and_then(serde_json::Value::as_str)
.filter(|arguments| !arguments.is_empty())
.map(str::to_string),
..Default::default()
}];
}
Ok(Some(event))
}
fn extract_responses_completed_tool_fragments(
parsed: &serde_json::Value,
) -> Result<Vec<ToolCallFragment>, LlmError> {
let Some(items) = parsed
.get("response")
.and_then(|response| response.get("output"))
.and_then(serde_json::Value::as_array)
else {
let Some(items) = extract_responses_output_array(parsed) else {
return Ok(Vec::new());
};
@@ -6320,6 +6376,90 @@ mod tests {
arguments: r#"{"city":"杭州"}"#.to_string(),
}]
);
assert_eq!(
response.responses_output,
vec![serde_json::json!({
"id":"fc_0",
"type":"function_call",
"status":"in_progress",
"arguments":"{\"city\":\"杭州\"}",
"call_id":"call_EkOU4",
"name":"get_weather"
})]
);
}
#[tokio::test]
async fn stream_run_replays_incremental_native_output_when_completed_omits_it() {
let server_url = spawn_mock_server(vec![MockResponse {
status_line: "200 OK",
content_type: "text/event-stream; charset=utf-8",
body: concat!(
r#"data: {"type":"response.output_item.added","item":{"type":"reasoning","id":"rs_1","summary":[],"encrypted_content":"encrypted-reasoning-payload"},"output_index":0}"#, "\n\n",
r#"data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","status":"in_progress","arguments":"","call_id":"call_1","name":"get_weather"},"output_index":1}"#, "\n\n",
r#"data: {"type":"response.function_call_arguments.done","item_id":"fc_1","output_index":1,"arguments":"{\"city\":\"杭州\"}"}"#, "\n\n",
r#"data: {"type":"response.output_item.done","item":{"id":"fc_1","type":"function_call","status":"completed","arguments":"{\"city\":\"杭州\"}","call_id":"call_1","name":"get_weather"},"output_index":1}"#, "\n\n",
r#"data: {"type":"response.completed"}"#, "\n\n"
)
.to_string(),
extra_headers: Vec::new(),
}]);
let response = build_test_client(server_url, 0)
.stream_run(
weather_tool_request(LlmApiKind::OpenAiResponses).with_responses_input(vec![
serde_json::json!({"role":"user", "content":"查询天气"}),
]),
|_| {},
)
.await
.expect("incremental native output");
assert_eq!(
response.responses_output,
vec![
serde_json::json!({
"type":"reasoning",
"id":"rs_1",
"summary":[],
"encrypted_content":"encrypted-reasoning-payload"
}),
serde_json::json!({
"id":"fc_1",
"type":"function_call",
"status":"completed",
"arguments":"{\"city\":\"杭州\"}",
"call_id":"call_1",
"name":"get_weather"
}),
]
);
assert_eq!(response.tool_calls[0].id, "call_1");
assert_eq!(response.tool_calls[0].arguments, r#"{"city":"杭州"}"#);
}
#[tokio::test]
async fn stream_run_uses_completed_top_level_output_when_response_wrapper_missing() {
let output = native_responses_output_fixture();
let body = format!(
"data: {}\n\n",
serde_json::json!({"type":"response.completed", "output":output})
);
let server_url = spawn_mock_server(vec![MockResponse {
status_line: "200 OK",
content_type: "text/event-stream",
body,
extra_headers: Vec::new(),
}]);
let response = build_test_client(server_url, 0)
.stream_run(
weather_tool_request(LlmApiKind::OpenAiResponses).with_responses_input(vec![
serde_json::json!({"role":"user", "content":"查询天气"}),
]),
|_| {},
)
.await
.expect("top-level completed output");
assert_eq!(response.responses_output, output);
assert_eq!(response.tool_calls[0].id, "call_1");
}
// 同一个终态载荷里两条相同 call_id、相同函数名、不同参数：上游违反了 call id 唯一性。