Agent 流式输出 —— 告别等待，让 LLM 边想边说

核心增量：stream_llm_call()（~80 行）+ run_agent_with_trace 调用点替换

一、问题：Agent 沉默了太久

用户要等 LLM 完整生成完才能看到任何输出，面对长分析报告或复杂推理时只能盯空白屏幕。更重要的是，当 LLM 决定调用工具时，用户完全不知道 Agent 在想什么——体验上就像一个黑盒。

流式输出要解决：让 LLM 像人一样"边想边说"，同时实时展示工具调用的决策过程。

二、核心原理：SSE 逐块推送

OpenAI 兼容 API 设置 stream=True 后，响应以 SSE（Server-Sent Events）格式逐块推送：

非流式                          流式
POST → 等待 → 完整响应           POST → chunk1 → chunk2 → ... → [DONE]
  └── 8 秒后拿到全文              └── 0.2s: "这段"
                                   ├── 0.4s: "代码"
                                   └── ... 逐字出现

关键：同一个 index 的 tool_call 可能跨越多个 chunk——name 在一个 chunk、arguments 在后续多个 chunk 中分批到达，需要手动拼接。

三、实现

def stream_llm_call(client, model, messages, tools):
    response = client.chat.completions.create(
        model=model, messages=messages, tools=tools,
        tool_choice="auto", stream=True,           # ← 关键
    )

    content_chunks = []
    tool_call_chunks = {}    # index → {id, name, arguments}
    shown_tool_names = set()

    print("🧠 ", end="", flush=True)

    for chunk in response:
        delta = chunk.choices[0].delta if chunk.choices else None
        if delta is None:
            continue

        if delta.content:                          # 文本：边收边打印
            print(delta.content, end="", flush=True)
            content_chunks.append(delta.content)

        if delta.tool_calls:                       # tool_calls：实时显示
            for tc_delta in delta.tool_calls:
                idx = tc_delta.index
                if idx not in tool_call_chunks:
                    tool_call_chunks[idx] = {"id": "", "name": "", "arguments": ""}
                if tc_delta.id:
                    tool_call_chunks[idx]["id"] = tc_delta.id
                if tc_delta.function.name:
                    tool_call_chunks[idx]["name"] = tc_delta.function.name
                    if idx not in shown_tool_names:
                        print(f"\n  🔧 调用 {tc_delta.function.name}", end="")
                        shown_tool_names.add(idx)
                if tc_delta.function.arguments:
                    tool_call_chunks[idx]["arguments"] += tc_delta.function.arguments

    if tool_call_chunks:
        tc_list = []
        for idx in sorted(tool_call_chunks.keys()):
            tc = tool_call_chunks[idx]
            tc_list.append({"id": tc["id"], "type": "function",
                            "function": {"name": tc["name"], "arguments": tc["arguments"]}})
        return None, tc_list
    else:
        print()
        return "".join(content_chunks), None

返回值设计

| 场景 | 返回 content | 返回 tool_calls | | --- | --- | --- | | LLM 直接回答 | "完整文本" | None | | LLM 要调工具 | None | [{id, function: {name, arguments}}] |

四、主循环集成

content, tool_calls = stream_llm_call(client, model, messages, active_tools)

if tool_calls is None:             # 纯文本答案
    return content

# 手动构造 assistant 消息（流式没有完整 msg 对象）
assistant_msg = {"role": "assistant", "content": content, "tool_calls": tool_calls}
messages.append(assistant_msg)

for tc in tool_calls:
    result = active_tool_map[tc["function"]["name"]](**json.loads(tc["function"]["arguments"]))
    messages.append({"role": "tool", "tool_call_id": tc["id"], "content": str(result)})

五、chunk 拼接的陷阱

一个 tool_call 的信息分散在多个 chunk，不能直接 append，要以 index 为 key 累加：

tool_call_chunks = {}  # index → 累加缓冲
for chunk in response:
    for tc_delta in delta.tool_calls:
        idx = tc_delta.index
        if idx not in tool_call_chunks:
            tool_call_chunks[idx] = {"id": "", "name": "", "arguments": ""}
        if tc_delta.id:
            tool_call_chunks[idx]["id"] = tc_delta.id
        if tc_delta.function.name:
            tool_call_chunks[idx]["name"] = tc_delta.function.name
        if tc_delta.function.arguments:
            tool_call_chunks[idx]["arguments"] += tc_delta.function.arguments

六、设计精要

返回值是二元组而非对象

流式响应没有原生 message 概念，二元组语义清晰——要么文本答案，要么工具调用列表。

工具名优先显示

一旦收到 function.name 就立刻打印 🔧 调用 xxx，不等到 arguments 收完。

Trace 的妥协

非流式能直接拿精确 token 数；流式下用 ctx.count_tokens + len(ctx.encoder.encode(...)) 估算。