这篇博客主要记录一下我在项目第二阶段做的工作：如何把原本“一次性蹦出来”的回复改成“打字机”效果，以及如何优化代码结构，让前后端逻辑分离得更清楚。

1. 为什么要搞流式输出？

在上一阶段，虽然我用“双阶段刷新”解决了发送消息后的白屏问题，但用户体验其实还是不够好。现在的 AI 回复是等后端全部生成完才一次性显示出来，如果回答很长，用户得盯着屏幕干等好几秒。

真正的 AI 应用（比如 ChatGPT）都是“流式”的——字是一个个蹦出来的。这样做有两个好处：一是让用户感觉系统反应很快（首字延迟低）；二是看着文字一点点生成，等待的焦虑感会少很多。

2. 代码结构的调整

为了实现流式，我不能再把模拟数据直接写在 app.py 里了。为了保持代码整洁，我把“怎么获取数据”和“怎么展示数据”分开了：

app_lib/streaming.py：专门负责“生产”数据。这里我写了一个生成器，用来模拟后端一点点吐出文字的过程。

app_lib/agent_client.py：这是一个中间层。它负责调用上面的生成器（未来这里会换成真实的 HTTP 请求），把数据拿回来。

app.py：只负责“展示”。它调用中间层拿到的数据流，然后用 Streamlit 的组件画出来。

通过这种分层，以后接入真实后端时，我只需要改 agent_client.py，完全不用动页面逻辑

3. 核心实现：模拟数据生成器

因为后端还没好，我得先自己造一个“假后端”。这里用到了 Python 的 yield 关键字，它能让函数变成一个“生成器”，每次只吐出一个字，然后暂停一下。

def _mock_reply_chunks(user_text: str) -> Iterator[str]:
    //模拟回复
    base = (
        "已收到您的需求，当前为**前端模拟流式**输出（未配置 AGENT_STREAM_URL）。\n\n"
        f"您提到：{user_text[:200]}{'…' if len(user_text) > 200 else ''}\n\n"
        "配置环境变量或 `.streamlit/secrets.toml` 中的 `AGENT_STREAM_URL` 后，"
        "将改为请求后端 SSE（`text/event-stream`，每行 `data:` 见 `sse_lines_to_text_chunks`）。"
    )
    step = max(1, MOCK_STREAM_CHUNK_CHARS)
    for i in range(0, len(base), step):
        yield base[i : i + step]
        time.sleep(0.012)

在这个阶段，我没有简单地按“字”来切分文本，而是设置了一个 step 步长（比如每次 8 个字符）。这样做是因为真实的网络流（SSE）通常是一次传输一小段数据块，而不是一个字一个字地发。通过 yield base[i : i + step]，我能更逼真地模拟出那种“数据一点点加载”的感觉。

4. 核心难点：Streamlit 的流式渲染

有了数据源，接下来就是怎么在页面上展示。Streamlit 提供了一个非常好用的组件叫 st.write_stream，它专门用来消费这种生成器数据。

相比起以前用 st.empty() 配合循环去不断更新文本，st.write_stream 更加丝滑。但我的 app.py 里不仅仅是简单的调用，我还处理了两个关键问题：

首字延迟优化（_prime_stream）：为了不让用户看到白屏，我写了一个 _prime_stream 包装器。它会先在 st.spinner 里预加载第一个数据包，确保连接建立后再把控制权交给 st.write_stream，这样能最大程度减少“转圈”的时间。

异常兜底：流式传输最怕断连。我用了 try...except 包裹了整个流式过程。如果连接失败，我会捕获 StreamConnectionError，并把已经接收到的部分内容（acc）显示出来，而不是让页面直接报错崩溃。

def _render_chat() -> None:
    st.title(f"{PAGE_ICON} {PAGE_TITLE}")

    if prompt := st.chat_input("描述区域、预算、户型等找房需求…"):
        append_message("user", prompt)
        st.session_state.pending_reply = True
        st.rerun()

    for msg in st.session_state.messages:
        with st.chat_message(msg["role"]):
            st.markdown(msg["content"])

    if st.session_state.get("pending_reply"):
        messages: list[dict[str, Any]] = st.session_state.messages
        user_text = ""
        if messages and messages[-1].get("role") == "user":
            user_text = str(messages[-1].get("content", ""))

        stream_url = resolve_stream_url()
        history = messages[:-1] if messages else []

        acc: list[str] = []
        sid = new_streaming_id()
        stream_error: str | None = None

        def _chunk_iter() -> Iterator[str]:
            for chunk in stream_assistant_reply(
                user_text,
                stream_url=stream_url,
                history=history,
            ):
                acc.append(chunk)
                yield chunk

        with st.chat_message("assistant"):
            try:
                st.write_stream(_prime_stream(lambda: _chunk_iter()))
            except StreamConnectionError as e:
                stream_error = str(e)
                st.error(stream_error)
                if acc:
                    st.caption("以下为已接收的部分内容：")
                    st.markdown("".join(acc))
            except Exception as e:  # noqa: BLE001 — 避免流中断时静默失败
                stream_error = str(e)
                st.error(f"流式输出异常：{e}")
                if acc:
                    st.markdown("".join(acc))

        full = "".join(acc)
        if not full.strip():
            full = (
                f"（响应失败：{stream_error}）"
                if stream_error
                else "（未收到有效回复，请检查后端或网络。）"
            )

        finished = time.time()
        append_message(
            "assistant",
            full,
            streaming_id=sid,
            finished_at=finished,
        )
        st.session_state.pending_reply = False
        st.rerun()

5. 总结与展望

现在的对话体验已经非常接近真实的 AI 产品了。字是一个个打出来的，而且因为我把数据获取逻辑封装在了 agent_client 里，整个 app.py 依然很干净。

接下来我要进入第三阶段：房源卡片与列表展示。