Fix live activity streaming and replay summary

This commit is contained in:
wuyang6
2026-05-13 20:58:35 +08:00
parent 0eb57f3f42
commit d7c62a4929
4 changed files with 59 additions and 4 deletions
+17 -1
View File
@@ -1901,7 +1901,14 @@ def create_app(state: AgentState) -> FastAPI:
break
yield json.dumps(item, ensure_ascii=False) + '\n'
return StreamingResponse(generate(), media_type='application/x-ndjson')
return StreamingResponse(
generate(),
media_type='application/x-ndjson',
headers={
'Cache-Control': 'no-cache, no-transform',
'X-Accel-Buffering': 'no',
},
)
@app.post('/api/clear')
async def clear_state(account_id: str | None = None) -> dict[str, Any]:
@@ -2117,12 +2124,17 @@ def _interrupted_session_message(status: str) -> str:
def _runtime_event_stage(event: dict[str, object]) -> str:
event_type = event.get('type')
if event_type == 'content_delta':
return ''
if event_type == 'tool_start':
stage_note = str(event.get('assistant_content') or '').strip()
if stage_note.startswith(('进度:', '进度:')):
return stage_note
tool_name = str(event.get('tool_name') or '').strip()
return f'调用工具 {tool_name}' if tool_name else '正在调用工具'
if event_type == 'tool_delta':
tool_name = str(event.get('tool_name') or '').strip()
return f'{tool_name} 输出中' if tool_name else '工具输出中'
if event_type == 'tool_result':
tool_name = str(event.get('tool_name') or '').strip()
return f'工具完成 {tool_name}' if tool_name else '工具调用完成'
@@ -2149,7 +2161,9 @@ def _runtime_event_stage(event: dict[str, object]) -> str:
_RUN_EVENT_TYPES = {
'run_queued',
'run_started',
'content_delta',
'tool_start',
'tool_delta',
'tool_result',
'final_text_start',
'final_text_end',
@@ -2175,8 +2189,10 @@ def _normalize_run_event(event: dict[str, object]) -> dict[str, Any] | None:
'tool_name',
'tool_call_id',
'arguments',
'delta',
'assistant_content',
'message_id',
'stream',
'ok',
'metadata',
'reason',
+13 -2
View File
@@ -56,6 +56,7 @@ type StreamState = {
phase: "waiting" | "queued" | "running" | "done";
runId?: string;
announcedToolCallKeys: Set<string>;
streamedToolCallIds: Set<string>;
};
type ChatRequestBody = {
@@ -112,6 +113,7 @@ export async function POST(req: Request) {
textStreamed: false,
phase: "waiting",
announcedToolCallKeys: new Set<string>(),
streamedToolCallIds: new Set<string>(),
};
writer.write({ type: "start" });
writer.write({ type: "start-step" });
@@ -142,7 +144,7 @@ export async function POST(req: Request) {
delta: `\n思考并执行完成,用时 ${formatDuration(elapsedMs)}`,
});
writer.write({ type: "reasoning-end", id: "reasoning-1" });
writeToolTrace(writer, payload.transcript);
writeToolTrace(writer, payload.transcript, streamState);
if (!streamState.textStreamed) {
writer.write({ type: "text-start", id: "text-1" });
writer.write({ type: "text-delta", id: "text-1", delta: text });
@@ -183,7 +185,13 @@ export async function POST(req: Request) {
},
});
return createUIMessageStreamResponse({ stream });
return createUIMessageStreamResponse({
stream,
headers: {
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
},
});
}
function formatDuration(ms: number) {
@@ -199,6 +207,7 @@ function formatDuration(ms: number) {
function writeToolTrace(
writer: UIMessageStreamWriter<UIMessage>,
transcript?: ClawTranscriptEntry[],
streamState?: StreamState,
) {
if (!transcript?.length) return;
@@ -216,6 +225,7 @@ function writeToolTrace(
const toolCallId = call.id;
const toolName = call.function?.name ?? call.name;
if (!toolCallId || !toolName) continue;
if (streamState?.streamedToolCallIds.has(toolCallId)) continue;
writer.write({
type: "tool-input-available",
@@ -539,6 +549,7 @@ function writeRuntimeEvent(
}
}
if (event.type === "tool_start" && event.tool_call_id && event.tool_name) {
streamState?.streamedToolCallIds.add(event.tool_call_id);
writer.write({
type: "tool-input-available",
toolCallId: event.tool_call_id,
@@ -979,6 +979,10 @@ function summarizeLiveRunEvents(
if (event.type === "run_started") {
lines.push("后端已开始执行本轮任务。");
}
if (event.type === "content_delta" && event.delta) {
const note = event.delta.trim();
if (note && !isNoisyLiveDelta(note)) lines.push(note);
}
if (event.type === "tool_start") {
const stageNote =
typeof event.assistant_content === "string"
@@ -994,6 +998,9 @@ function summarizeLiveRunEvents(
event.tool_name ? `工具完成 ${event.tool_name}` : "工具调用完成",
);
}
if (event.type === "tool_delta") {
lines.push(event.tool_name ? `${event.tool_name} 输出中` : "工具输出中");
}
if (event.type === "final_text_start") {
lines.push("正在整理回复");
}
@@ -1011,6 +1018,10 @@ function summarizeLiveRunEvents(
return lines.slice(-40);
}
function isNoisyLiveDelta(value: string) {
return value === "。" || value === "" || value === "," || value.length <= 1;
}
function collectActivityItems(messages: readonly MessageState[]) {
const items: ActivityItem[] = [];
@@ -86,6 +86,7 @@ type ClawRunEvent = {
tool_call_id?: string;
arguments?: unknown;
assistant_content?: string;
delta?: string;
ok?: boolean;
metadata?: {
output_preview?: unknown;
@@ -430,7 +431,12 @@ export function toReplayRepository(
? shouldShowAssistantText
? [...toolParts, { type: "text", text: content }]
: toolParts
: [{ type: "text", text: content }]
: message.role === "assistant" && elapsedMs !== undefined
? [
{ type: "reasoning", text: "思考并执行完成。", elapsedMs },
{ type: "text", text: content },
]
: [{ type: "text", text: content }]
) as UIMessage["parts"];
const replayParts = withReplayElapsedMs(parts, elapsedMs);
const uiMessage: UIMessage = {
@@ -599,6 +605,10 @@ function summarizeRunEvents(runStatus: ClawActiveRunStatus) {
if (event.type === "run_started") {
lines.push("后端已开始执行本轮任务。");
}
if (event.type === "content_delta" && event.delta) {
const note = event.delta.trim();
if (note && !isNoisyLiveDelta(note)) lines.push(note);
}
if (event.type === "tool_start") {
const stageNote =
typeof event.assistant_content === "string"
@@ -614,6 +624,9 @@ function summarizeRunEvents(runStatus: ClawActiveRunStatus) {
event.tool_name ? `工具完成 ${event.tool_name}` : "工具调用完成",
);
}
if (event.type === "tool_delta") {
lines.push(event.tool_name ? `${event.tool_name} 输出中` : "工具输出中");
}
if (event.type === "final_text_start") {
lines.push("正在整理回复");
}
@@ -627,6 +640,10 @@ function summarizeRunEvents(runStatus: ClawActiveRunStatus) {
return lines.slice(-40);
}
function isNoisyLiveDelta(value: string) {
return value === "。" || value === "" || value === "," || value.length <= 1;
}
function buildRunToolParts(events: readonly ClawRunEvent[]) {
const resultById = new Map<string, ClawRunEvent>();
for (const event of events) {