本文摘要回复一次性返回要等生成完并发共用 history 会互相串台。新增按增量产出的流式方法用 SSE 逐块推送长回复。单客户端能逐字输出但 history 是共享单例并发前须隔离会话。一、环境与前提上一篇六用api.py的 FastAPI 接口把agent-demo/里的智能体包成了 HTTP 服务回复只能一次性返回本篇给它补上流式输出并处理并发请求下history的归属。目的确认起点仍是第 5、6 篇的产出本篇只新增两个导入不引入新的第三方依赖。第 5 篇的Agent、history维护、_count_tokens、_truncate_messages与第 6 篇的ChatRequest、ChatResponse、chat、get_agent一律沿用原写法环境安装与请求校验不再重复展开。操作进入项目目录做一次语法基线检查。工作目录必须是agent-demo/运行环境沿用系列设定 Python 3.11、openai 1.xjson是标准库StreamingResponse来自已安装的fastapi两者都不需要重新安装。cdagent-demo python-mpy_compile main.py api.py预期输出命令无任何输出退出码为 0说明第 6 篇结束时的两个文件语法完整。实际输出未实测。二、关键步骤步骤 1在main.py的Agent中新增run_agent_stream目的让回复按增量产出不再等run_agent把整段文本攒完history的维护与_truncate_messages截断仍沿用第 5 篇写法只换产出方式。操作把下面的方法粘进Agent类内、run_agent方法之后缩进与run_agent一致4 空格。self.client、self.model、self.history三个属性沿用第 5 篇__init__里的原名不要改名defrun_agent_stream(self,user_message):与 run_agent 共用同一套 history 维护只是把回复按增量 yield 出来。self.history.append({role:user,content:user_message})messagesself._truncate_messages(self.history)reply[]forchunkinself.client.chat.completions.create(modelself.model,messagesmessages,streamTrue,):deltachunk.choices[0].delta.contentifnotdelta:continuereply.append(delta)yielddelta self.history.append({role:assistant,content:.join(reply)})两处关键点streamTrue让create(...)返回可迭代的分块对象增量文本取自chunk.choices[0].delta.content部分块该字段为None用if not delta: continue跳过依据 openai-python 库文档的 Streaming 示例openai 1.x小版本未确认结尾那行把整段回复补回history否则下一轮_truncate_messages(self.history)里没有助手侧内容。结果在agent-demo/目录做语法检查。python-mpy_compile main.py预期输出命令无任何输出退出码为 0。实际输出未实测。步骤 2在api.py新增chat_stream端点目的把增量块以text/event-stream写回连接调用方可以边收边渲染原chat端点与ChatResponse原样保留两个接口并存。操作先在api.py顶部补两行导入import json放进标准库导入区另一行与fastapi的导入放在一起importjsonfromfastapi.responsesimportStreamingResponse再在chat端点之后追加下面的端点函数ChatRequest、get_agent用第 6 篇的原写法request.message是ChatRequest里的用户输入字段app.post(/chat/stream)defchat_stream(request:ChatRequest):agentget_agent()defgenerate():fordeltainagent.run_agent_stream(request.message):yieldfdata:{json.dumps({delta:delta},ensure_asciiFalse)}\n\nyielddata: [DONE]\n\nreturnStreamingResponse(generate(),media_typetext/event-stream)StreamingResponse接受生成器作为响应体按块写回连接media_type由调用方指定依据 FastAPI 自定义响应文档版本未确认。每条事件以data:开头、空行分隔末尾固定发一个data: [DONE]客户端据此判断流结束状态码在流开始时已经发出之后的信息只能靠事件传递。结果在agent-demo/目录做语法检查。python-mpy_compile api.py预期输出命令无任何输出退出码为 0。实际输出未实测。步骤 3启动服务并自测两个接口目的确认增量块真的逐块到达且旧的POST /chat没被破坏。操作终端 A 在agent-demo/目录启动服务启动方式与第 6 篇相同uvicorn api:app--reload终端 B 发起流式请求-N用于关掉 curl 的输出缓冲否则小块响应会攒在终端里不显示curl-N-XPOST http://127.0.0.1:8000/chat/stream\-HContent-Type: application/json\-d{message: 用一句话解释什么是流式输出}预期输出终端按块打印data:开头、空行分隔的事件行每行形如data: {delta: 增量文本}增量文本由模型生成、内容不可预测最后一行固定是data: [DONE]。实际输出未实测。接着回归旧接口在同一终端再执行curl-XPOST http://127.0.0.1:8000/chat\-HContent-Type: application/json\-d{message: 你好}预期输出一次性返回一个 JSON 对象字段与第 6 篇的ChatResponse完全一致。实际输出未实测。到这一步agent-demo/处于完整可运行状态POST /chat返回第 6 篇的 JSON 响应POST /chat/stream返回增量事件流两者共用同一个Agent实例与同一份history。三、失败处理坑 1在流式块上照搬非流式字段触发把非流式返回里的choices[0].message.content取值方式直接搬到run_agent_stream的循环里写成chunk.choices[0].message.content。报错原文AttributeError: Choice object has no attribute messagePython 3.11 下这条异常后面可能还会附带一句Did you mean: ...提示核心文本不变。原因流式分块对象里的候选项类型与非流式不同只带delta一类的增量字段没有message字段依据 openai-python 库文档的 Streaming 示例openai 1.x。修复改取chunk.choices[0].delta.content并对None跳过也就是步骤 1 代码里的那两行。验证重跑步骤 3 的 curl 命令。预期输出回到分块事件行终端里不再出现AttributeError。实际输出未实测。坑 2把整个分块对象丢给json.dumps触发写成yield fdata: {json.dumps(chunk)}\n\n想把完整块转发给客户端。报错原文TypeError: Object of type ChatCompletionChunk is not JSON serializable原因分块对象不是json模块可序列化的标准类型json.dumps遇到这类对象抛TypeError依据 Pythonjson模块文档Python 3.11。修复只取文本增量再序列化写成json.dumps({delta: delta}, ensure_asciiFalse)即步骤 2 代码里的写法。验证重跑步骤 3 的 curl 命令。预期输出除结束标记data: [DONE]外每个事件行的负载都是合法 JSON结构统一为{delta: ...}。实际输出未实测。坑 3流式回复没有写回history不报错但结果错现象第一轮对话正常第二轮提问时智能体失忆。原因run_agent_stream是分块产出的如果只yield delta而不把完整回复拼回history下一轮就没有助手侧内容第 5 篇建立的对话历史维护被绕过。修复用reply列表收集增量在生成器结束前补上self.history.append({role: assistant, content: .join(reply)})也就是步骤 1 代码的最后一行。验证连续两轮调用POST /chat/stream第二轮问「我上一句问的是什么」。预期输出第二轮回复能引用第一轮的提问内容。实际输出未实测回复正文由模型生成无法预写。坑 4并发请求共用单例history不报错但数据互相污染现象两个客户端同时打POST /chat/stream回复里混进对方的上文。原因同步def端点在每个请求一个独立线程里执行而get_agent()返回的是同一个Agenthistory是共享可变列表依据 FastAPI「Concurrency and Burgers」一节版本未确认。修复方式有三种见下一节表二。验证复现两个终端同时执行步骤 3 的 curl 命令。预期输出按表二选项 A 的现状实现两份回复里会混进对方的上文具体文本由模型生成无法预写。实际输出未实测。四、替代方案与取舍表一流式 SSE 与非流式 JSON 的取舍。两者不互斥本篇让它们并存调用方按场景选。做法适用条件代价边界非流式 JSON第 6 篇的chat回复短、调用方是脚本或需要一次性校验与落库首字延迟等于全量生成时间长回复等待明显复用第 6 篇的ChatResponse改动最少出错仍可用 HTTP 状态码区分流式 SSE本篇的chat_stream回复长、有人盯着屏幕等字客户端要自己拼接增量块多出[DONE]约定的维护成本错误只能在流里以事件告知日志与缓存更难做连接中断时已生成内容无法撤回对一次性短回复收益趋近于零表二并发下history的归属。同步端点每请求一线程必须在下面三种里选一种。做法适用条件代价边界A. 单例共享一份history现状本地单人调试、串行调用无需改动只要并发大于 1 就会上下文互染B. 会话隔离ChatRequest追加可选session_id带默认值旧请求不报错get_agent()换成dict加threading.Lock多用户或多任务的同进程服务内存随会话数增长需要过期与清理策略进程重启会话即失效跨进程多副本仍不成立C. 无状态请求体携带历史服务端不保存多副本部署、需要水平扩展请求体变大token 计费与截断压力移到每次请求客户端回传不全仍会失忆第 5 篇的history维护责任移出进程边界要与调用方约定只作讨论、本篇不实现的方向把chat_stream改成async def加异步生成器async def里yield即异步生成器函数并发结构更清晰但 openai 的调用要换成异步客户端否则一次阻塞调用会拖住事件循环依据 FastAPI「Concurrency and Burgers」与 Python 异步生成器函数一节。这涉及替换第 5 篇Agent内部的调用写法超出本篇范围。选型落到具体场景回复短、调用方是脚本或需要一次性校验落库就用第 6 篇的chat要多副本部署或横向扩展就直接选表二的选项 C不要把history继续留在进程内单例里。参考资料OpenAI Streaming Responses 指南openai 1.x小版本未确认FastAPI Custom Response 之 StreamingResponse版本未确认页面未标注发布时间FastAPI Concurrency and Burgers版本未确认页面未标注发布时间Python 复合语句之异步生成器函数Python 3.11Python json 模块文档Python 3.11下一篇在chat_stream之上补做并发请求下的会话隔离与超时控制让多客户端同时使用时上下文不再互染。