教程：https://www.bilibili.com/video/BV1yjz5BLEoY

代码：https://github.com/shangxiang0907/HeiMa-AI-LLM-RAG-Agent-Dev

云开发平台：https://hzh.sealos.run

Agent智能体-13、[Agent项目]用户界面开发

介绍

→提示词：我现在要实现@zhisaotong_agent/PROJECT_OVERVIEW.md 中的app.py，下面是一个已有实现，请指出它在生产环境下的不足

import time

import streamlit as st
from agent.react_agent import ReactAgent

# 标题
st.title("智扫通机器人智能客服")
st.divider()

if "agent" not in st.session_state:
    st.session_state["agent"] = ReactAgent()

if "message" not in st.session_state:
    st.session_state["message"] = []

for message in st.session_state["message"]:
    st.chat_message(message["role"]).write(message["content"])

# 用户输入提示词
prompt = st.chat_input()

if prompt:
    st.chat_message("user").write(prompt)
    st.session_state["message"].append({"role": "user", "content": prompt})

    response_messages = []
    with st.spinner("智能客服思考中..."):
        res_stream = st.session_state["agent"].execute_stream(prompt)

        def capture(generator, cache_list):

            for chunk in generator:
                cache_list.append(chunk)

                for char in chunk:
                    time.sleep(0.01)
                    yield char

        st.chat_message("assistant").write_stream(capture(res_stream, response_messages))
        st.session_state["message"].append({"role": "assistant", "content": response_messages[-1]})
        st.rerun()

←AI回复

你这段 app.py 在“能跑”的 demo 形态下没问题，但放到真实生产（多用户并发、长时间运行、需要稳定可观测）会暴露一些明显不足，主要集中在 正确性、性能、可靠性/运维 三块：

1) 对话内容记录存在严重正确性问题

• 你把流式结果分块累到 response_messages 里，但最终写入历史的方式是：
  • st.session_state["message"].append({"role": "assistant", "content": response_messages[-1]})
• 这会导致：“assistant 的历史消息只保存最后一个 chunk”，而不是完整回答（下一次 st.session_state 重绘时就会显示被截断的内容）。
• 另外还有边界风险：如果模型流式过程中没有产出任何 chunk（极端情况/异常分支），response_messages[-1] 会直接触发 IndexError，让整个会话崩掉。

2) 流式渲染的性能与吞吐会被额外放大成本

• 你在 capture() 里对每个字符都 time.sleep(0.01)：
  • for char in chunk: time.sleep(0.01); yield char
• 这属于“人为降低吞吐”，生产环境会造成：
  • 同一 worker 上被阻塞更久，并发量上来后会堆积请求；
  • 增加用户感知延迟；
  • 在模型本身已经是流式输出的情况下，完全没必要再逐字符 sleep（会让流式优势消失）。

3) 缺少关键的初始化与配置校验（可能直接导致线上失败）

• ReactAgent() 内部会调用 get_chat_model()（来自 model/factory.py），而 get_chat_model() 不会替你初始化 API Key。
• init_dashscope_api_key() 只在 react_agent.py 的 __main__ 自测入口里调用过；你的 app.py 没有调用它。
• 结果就是：生产环境如果没有保证启动时环境变量已就绪，首次创建 ReactAgent() 就可能失败（至少会表现为模型初始化异常/不可用），但你当前 app.py 没有做任何错误提示或降级。

4) 缺少异常处理与可观测性兜底

• 你对 res_stream = st.session_state["agent"].execute_stream(prompt) 及后续流式消费没有 try/except。
• 线上一旦模型超时、网络抖动、RAG 工具报错、LLM SDK 抛异常，你的页面会直接报错或静默中断，用户不知道发生了什么。
• 虽然 middleware.py 里有日志/脱敏/监控能力（monitor_tool/log_before_model/report_prompt_switch），但那只覆盖“agent 内部链路”，UI 层仍需要捕获异常并给用户友好反馈 + 记录失败上下文。

5) 并发/会话层面的资源与状态治理不足

• 你用 st.session_state["agent"] = ReactAgent()：每个用户会话都会创建一个 agent 实例（虽然底层 chat model 是全局懒加载单例，但 agent/链路对象本身仍是按会话创建）。
  • 在生产并发上来时，创建成本可能抬升启动延迟与内存占用。
• st.session_state["message"] 没有限制长度。
  • 长对话会导致 session 内存持续增长，并且每次重跑脚本要渲染全部历史消息，越来越慢。

6) st.rerun() 的时机不够“稳健”

• 你在流式输出结束后强制 st.rerun()。
• 虽然你通过 prompt = st.chat_input() 的语义通常不会重复触发同一次请求，但生产环境里仍会带来：
  • 额外一次完整脚本重跑（浪费算力）；
  • 与“上面提到的 assistant 历史截断问题”叠加后，用户看到的最终历史会被错误覆盖。

---

如果你愿意，我也可以基于你这份代码给出“最小修改版”的生产化改法（例如：如何正确拼接完整流输出、如何处理空流、如何初始化 API Key、如何加超时/异常兜底、以及如何去掉逐字符 sleep 并保持稳定的流式体验）。

→提示词：帮我实现你的修改

←AI回复

已按之前的结论在项目里新增并接好 zhisaotong_agent/app.py，主要改动如下。

实现说明

1. 启动时校验 API Key
   每个会话只调用一次 init_dashscope_api_key() 并写入 st.session_state["_dashscope_ok"]；失败则 st.error + st.stop()，避免在缺 Key 时再去构造 ReactAgent。

2. 智能体初始化
   ReactAgent() 包在 try/except 里，失败会打日志并提示用户，避免裸异常把整页打挂。

3. 完整保存助手回复
   流式过程中用列表收集所有 chunk，"".join(...).strip() 后再写入 messages，不再只用 response_messages[-1]。

4. 去掉逐字 time.sleep
   直接 yield 模型返回的 chunk，不再人为拖慢吞吐。

5. 去掉 st.rerun()
   依赖 Streamlit 正常重跑即可；历史由顶部的 for msg in messages 统一渲染。

6. 异常与空结果

   • execute_stream 异常：logger.exception + st.error，并写入固定失败文案，避免 IndexError。
   • 流结束但拼接后为空：写入占位说明文案。

7. 会话长度上限
   _MAX_SESSION_MESSAGES = 200，超出从头部删除，减轻 session 与首屏渲染压力。

8. 页面与文案
   st.set_page_config、中文省略号与占位提示与产品名一致。

运行方式（仓库根目录）：

streamlit run zhisaotong_agent/app.py

若你希望失败时保留「已生成的一半」再拼接说明，或要把 200 改成读 config/agent.yml，可以说明一下我再加一版。

代码

app.py

"""
智扫通 Streamlit 入口：会话展示、流式回复与基础生产化防护。

在 ``zhisaotong_agent/`` 目录执行 ``pip install -e .`` 后，从仓库根目录启动（推荐用 ``python3 -m``，不依赖 ``streamlit`` 是否在 PATH）::

    python3 -m streamlit run zhisaotong_agent/src/zhisaotong_agent/app.py

若在 ``zhisaotong_agent/`` 目录下，则::

    python3 -m streamlit run src/zhisaotong_agent/app.py
"""

from __future__ import annotations

import streamlit as st

from zhisaotong_agent.agent.react_agent import ReactAgent
from zhisaotong_agent.utils.api_key import init_dashscope_api_key
from zhisaotong_agent.utils.logger_handler import get_logger

logger = get_logger(__name__)

# 会话消息条数上限（user/assistant 各算一条），防止长时间对话撑爆 session 与首屏渲染
_MAX_SESSION_MESSAGES = 200


def _trim_messages(messages: list[dict[str, str]], max_items: int) -> None:
    while len(messages) > max_items:
        messages.pop(0)


st.set_page_config(page_title="智扫通机器人智能客服", layout="centered")
st.title("智扫通机器人智能客服")
st.divider()

if "_dashscope_ok" not in st.session_state:
    st.session_state["_dashscope_ok"] = init_dashscope_api_key()

if not st.session_state["_dashscope_ok"]:
    st.error(
        "未检测到有效的 DASHSCOPE_API_KEY（或 API_KEY）。"
        "请在环境变量或项目根目录 .env 中配置后刷新页面。"
    )
    st.stop()

if "agent" not in st.session_state:
    try:
        st.session_state["agent"] = ReactAgent()
    except Exception:
        logger.exception("ReactAgent 初始化失败")
        st.error("智能体初始化失败，请检查配置文件与依赖服务是否正常。")
        st.stop()

if "messages" not in st.session_state:
    st.session_state["messages"] = []

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

prompt = st.chat_input("请输入您的问题…")

if prompt:
    st.session_state["messages"].append({"role": "user", "content": prompt})
    _trim_messages(st.session_state["messages"], _MAX_SESSION_MESSAGES)

    with st.chat_message("user"):
        st.markdown(prompt)

    collected: list[str] = []

    def _stream() -> str:
        for chunk in st.session_state["agent"].execute_stream(prompt):
            collected.append(chunk)
            yield chunk

    assistant_text: str
    with st.chat_message("assistant"):
        with st.spinner("智能客服思考中…"):
            try:
                st.write_stream(_stream())
            except Exception:
                logger.exception("execute_stream 失败, prompt_len=%s", len(prompt))
                st.error("抱歉，当前请求处理失败，请稍后重试。")
                assistant_text = "（本次回复生成失败，请重试。）"
            else:
                assistant_text = "".join(collected).strip()
                if not assistant_text:
                    assistant_text = "（未返回有效内容，请换种方式提问或稍后重试。）"

    st.session_state["messages"].append({"role": "assistant", "content": assistant_text})
    _trim_messages(st.session_state["messages"], _MAX_SESSION_MESSAGES)

运行

cd zhisaotong_agent
python -m pip install -e .

python3 -m streamlit run zhisaotong_agent/src/zhisaotong_agent/app.py --server.address 0.0.0.0 --server.port 8501

结果