可审计的ReAct Agent：SSE流式可视化与全栈可观测性实现

1. 这不是又一个“前后端分离”Demo为什么ReAct Agent必须可审计、可追溯、可调试你有没有试过这样的情形写好一个调用大模型的Agent跑起来能回答问题、能执行工具、甚至能写代码——但一旦出错你完全不知道它在想什么。它跳过了本该调用的数据库查询却突然去调用了天气API它声称“已确认用户意图”但日志里只有一行{status: success}你改了提示词结果整个决策链路全乱了连重放一次都做不到。这就是典型的“黑盒Agent”困境。而标题里这个项目——AI 全栈学习之旅 - Week 11从 CLI 到浏览器用 FastAPI、SSE 和 Vue 3 做一个可审核的 ReAct Agent——它的核心价值根本不在“用了FastAPI”或“用了Vue 3”而在于把ReAct范式中那个被长期忽略的“思考过程”Thought真正变成可观察、可截断、可回溯的一等公民。关键词里没有出现“audit”“trace”“log”“step-by-step”但摘要描述里那句“可审核的ReAct Agent”才是题眼。它直指当前AI应用开发中最痛的盲区我们花大量精力优化prompt、选模型、搭pipeline却对Agent内部如何推理、何时失败、在哪一步卡住毫无掌控力。我带过6个AI工程化落地项目其中4个在上线后两周内因“无法定位Agent异常行为”被迫降级为静态问答。不是模型不行是整个链路缺乏可观测性设计。而这个Week 11项目用一套轻量但完整的全栈结构把ReAct的Thought → Action → Observation → … → Answer每一步都变成浏览器里可展开、可筛选、可导出的结构化事件流。它不依赖任何商业可观测平台不引入复杂中间件就靠FastAPI的SSE流、Vue 3的响应式状态管理、以及CLI端对标准输入输出的精准控制把“思考过程”从日志文件里解放出来变成前端可交互的实时面板。适合谁看如果你正在用LangChain/LlamaIndex写Agent但每次debug都要翻10个日志文件在Vue/React项目里嵌入AI能力却只能显示“加载中…”和最终答案想给非技术同事演示Agent工作原理但PPT里的流程图永远和实际运行对不上或者只是想搞懂SSE到底怎么让浏览器“看到”后端每一步思考而不是等全部算完才吐一个JSON——那你就是这个项目的理想读者。它不教FastAPI怎么写路由不讲Vue 3的Composition API原理也不深挖ReAct的学术定义。它只做一件事用最小技术栈把Agent的“大脑活动”实时、保真、无损地映射到用户界面上。接下来所有内容都围绕这个目标展开。2. 为什么必须用SSE而不是WebSocket或轮询流式思考的底层契约很多初学者看到“实时展示Agent步骤”第一反应是WebSocket。毕竟它双向、低延迟、听着就很“高级”。但在这个项目里选择SSEServer-Sent Events不是妥协而是基于ReAct Agent工作模式的精准匹配。让我用一个真实对比说明假设Agent要完成“查北京天气→如果温度低于15℃→推荐穿羽绒服”任务。WebSocket方案前端发{ query: 北京天气 }→ 后端启动ReAct循环 → 每步思考Thought/Action/Observation都通过socket.send()推给前端 → 最终socket.send({ answer: 穿羽绒服 })。SSE方案前端用EventSource连接/api/agent/stream→ 后端用yield逐行推送data: {step: thought, content: 需要获取北京当前天气...}→data: {step: action, tool: weather_api, params: {city: 北京}}→ … →data: {step: answer, content: 穿羽绒服}。表面看差不多关键差异在语义契约和错误恢复机制上。2.1 SSE的天然优势单向流 自动重连 文本协议ReAct Agent的本质是单向推理流后端驱动前端只消费。它不需要前端随时中断Agent比如“停别查天气了”也不需要前端主动推送新状态比如“用户刚点了‘再想想’按钮”。这种“后端推、前端收”的模式正是SSE的设计哲学。而WebSocket强制要求双向通道徒增复杂度——你得设计消息类型、处理ACK、管理连接状态最后发现90%的代码都在维护连接而非展示思考。更关键的是SSE的自动重连机制。当Agent处理耗时较长比如调用外部API等待3秒网络抖动导致连接断开SSE客户端会自动尝试重连并带上上次收到的Last-Event-ID。后端只需检查该ID从断点继续推送后续步骤。而WebSocket断开后前端必须手动重建连接、重新发起请求、并自行维护断点续传逻辑——这对一个“展示思考过程”的功能来说纯属冗余负担。再看协议层面。SSE是纯文本协议每条消息格式固定event: step data: {step: thought, content: 需要获取北京当前天气...} id: 12345 event: action data: {step: action, tool: weather_api, params: {city: 北京}} id: 12346这种结构让前端解析极其简单监听step事件JSON.parse(event.data)即可。而WebSocket传输二进制或JSON字符串你需要自己约定分隔符、处理粘包、校验消息完整性。在Vue 3里一行const eventSource new EventSource(/api/agent/stream)就能搞定而WebSocket需要new WebSocket()onopenonmessageonerror 重连逻辑代码量翻倍且易出错。2.2 那些“stream disconnected before completion: idle timeout waiting for sse”的真实原因网络热词里高频出现的stream disconnected before completion: idle timeout waiting for sse绝不是SSE本身的问题而是开发者忽略了HTTP连接的生命周期管理。SSE本质是长连接HTTP请求服务器必须持续发送数据哪怕心跳否则Nginx/Apache/Cloudflare等代理层会在60秒左右主动断开空闲连接。我在PyCharm调试时踩过这个坑本地FastAPI用uvicorn.run()启动默认timeout_keep_alive5秒。当Agent卡在某个API调用比如天气接口超时后端没及时推送新消息Uvicorn认为连接空闲直接关闭socket。浏览器报错NetworkError when attempting to fetch resource然后疯狂重连形成雪崩。解决方案不是调大timeout而是主动发送心跳# fastapi_backend/main.py from fastapi import Response import asyncio import json app.get(/api/agent/stream) async def stream_agent_response(query: str): async def event_generator(): # 初始化Agent获取思考流 agent_stream get_react_agent_stream(query) # 发送初始事件告诉前端开始 yield fevent: init\ndata: {json.dumps({status: started})}\n\n # 主循环逐个yield步骤 async for step in agent_stream: yield fevent: step\ndata: {json.dumps(step)}\n\n # 关键每5秒发一次心跳防止代理层断连 await asyncio.sleep(0.001) # 防止阻塞 # 结束事件 yield fevent: done\ndata: {json.dumps({status: completed})}\n\n return Response( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, } )注意两点一是await asyncio.sleep(0.001)不是为了延时而是让异步循环有机会调度避免长时间阻塞二是headers里明确声明no-cache和keep-alive这是SSE的黄金配置。很多教程漏掉这点导致生产环境必现超时。2.3 为什么不用轮询Polling轮询看似简单前端setInterval(() fetch(/api/agent/status), 1000)。但它有不可接受的缺陷延迟高最坏情况要等999ms才看到新步骤资源浪费Agent只产生10步你却发了100次请求状态不一致第5步和第6步可能被两个不同请求拿到前端需自行合并无法表达“结束”你永远不知道Agent是否真的完成了还是刚好卡在两次轮询之间。SSE用一个连接承载所有事件天然解决这些问题。它不是“更简单”而是“更正确”。3. CLI端不只是命令行入口而是ReAct Agent的标准化输入/输出契约项目标题里“从 CLI 到浏览器”中的CLI常被误解为“给开发者用的调试工具”。实际上在这个架构里CLI是ReAct Agent与外部世界交互的标准化边界。它不负责实现Agent逻辑只定义输入格式、输出协议、错误码体系——就像USB接口不管里面是U盘还是摄像头插上去就得按USB协议说话。3.1 CLI的核心职责解耦Agent实现与交互方式想象一下你的ReAct Agent后端用Python写前端用Vue 3但某天产品说“要支持微信小程序”。如果后端直接暴露HTTP接口给小程序你得为小程序单独写一套鉴权、重试、错误提示逻辑。而有了CLI层小程序只需调用codex-cli run --query 北京天气CLI负责将--query参数转成标准JSON请求体处理HTTP超时、重试、token刷新解析SSE流按event类型分类输出thought打蓝字action打黄字error打红字当收到done事件返回标准退出码0成功1用户错误2系统错误。这样Agent后端完全不用关心调用方是谁。CLI成了“协议转换器”让同一套Agent能力既能被浏览器调用也能被Shell脚本、CI/CD流水线、甚至另一个Agent调用。3.2 实现一个生产级CLIClick Rich Typer的组合拳网络热词里反复出现unable to locate the codex cli binary or required runtime components这暴露了一个常见误区把CLI当成python main.py的包装。真正的CLI必须是自包含的可执行文件用户下载即用不依赖其本地Python环境。我们用PyInstaller打包但构建前需精心设计CLI结构。核心依赖选型Typer比Click更现代原生支持异步命令、自动补全、OpenAPI文档生成Rich终端渲染库让思考步骤带颜色、进度条、折叠面板远超print()httpx异步HTTP客户端完美匹配SSE流式读取。CLI主文件cli.py结构如下# cli/codex_cli.py import typer from rich.console import Console from rich.panel import Panel from rich.text import Text import httpx import asyncio import json console Console() app typer.Typer( namecodex-cli, helpA CLI for interacting with ReAct Agents, add_completionFalse, ) app.command() def run( query: str typer.Argument(..., helpThe user query to process), base_url: str typer.Option(http://localhost:8000, --base-url, helpBackend API base URL), ): Run a ReAct Agent query and stream steps in terminal. console.print(Panel(f[bold]Starting ReAct Agent for:[/bold] {query}, expandFalse)) async def _stream_steps(): async with httpx.AsyncClient() as client: try: async with client.stream(GET, f{base_url}/api/agent/stream?query{query}) as response: if response.status_code ! 200: console.print(f[red]Error:[/red] Backend returned {response.status_code}) return async for line in response.aiter_lines(): if line.strip() : continue if line.startswith(data:): try: data json.loads(line[5:].strip()) _render_step(data) except json.JSONDecodeError: pass # 忽略无效JSON except httpx.ConnectError: console.print([red]Connection failed. Is backend running?[/red]) except Exception as e: console.print(f[red]Unexpected error:[/red] {e}) asyncio.run(_stream_steps()) def _render_step(step_data: dict): Render a single ReAct step with Rich formatting. step_type step_data.get(step, unknown) content step_data.get(content, ) if step_type thought: text Text( Thought:, stylebold blue) text.append(f {content}) console.print(text) elif step_type action: tool step_data.get(tool, unknown) params step_data.get(params, {}) text Text(️ Action:, stylebold yellow) text.append(f {tool}({params})) console.print(text) elif step_type observation: text Text( Observation:, stylebold green) text.append(f {content[:100]}{... if len(content) 100 else }) console.print(text) elif step_type answer: text Text(✅ Answer:, stylebold green) text.append(f {content}) console.print(Panel(text, border_stylegreen)) elif step_type error: text Text(❌ Error:, stylebold red) text.append(f {content}) console.print(Panel(text, border_stylered)) if __name__ __main__: app()打包命令# 安装依赖 pip install typer rich httpx pyinstaller # 打包为单文件 pyinstaller --onefile --name codex-cli cli/codex_cli.py # 生成的 ./dist/codex-cli 就是用户可直接运行的二进制3.3 CLI的“可审核性”体现在哪可审核性不是加个日志就完事。CLI通过三个设计保障审计价值结构化输出每步都带step类型标签方便grep或ELK收集。比如codex-cli run 北京天气 21 | grep thought直接提取所有思考步骤。时间戳注入在_render_step里加入console.print(f[dim]{datetime.now().isoformat()}[/dim])所有输出自带毫秒级时间戳跨服务追踪成为可能。退出码语义化sys.exit(0)表示Agent正常完成sys.exit(1)表示用户输入错误如空querysys.exit(2)表示后端不可达。运维脚本可据此自动告警无需解析文本。这才是CLI作为“审计锚点”的意义它把模糊的“Agent运行了”变成精确的“Agent在14:23:05.123启动14:23:07.456完成共12步其中3步为Observation”。4. Vue 3前端用响应式状态管理重构“思考过程”的可视化范式很多教程教Vue 3调用SSE最后只做一个div{{ currentStep }}/div。这完全浪费了Vue 3的响应式能力。在这个项目里Vue 3不是“展示层”而是ReAct Agent思考过程的实时状态镜像。我们用Composition API Pinia把每一步思考建模为可响应、可操作、可持久化的状态对象。4.1 状态设计为什么不用数组push而要用Mapref初学者常这样写// ❌ 错误示范简单数组 const steps ref([]) const eventSource new EventSource(/api/agent/stream) eventSource.onmessage (e) { steps.value.push(JSON.parse(e.data)) }问题在于数组push触发整个列表重渲染100步思考会导致100次DOM更新卡顿无法按类型筛选比如只看action步骤无法随机访问某步比如点击第5步展开详情无法标记某步为“已验证”审计需要人工确认。正确做法是用Map存储ref包裹// ✅ 正确Map ref computed import { ref, computed, onUnmounted } from vue import { defineStore } from pinia export const useAgentStore defineStore(agent, () { // MapstepId, StepObjectstepId由后端生成如UUID或递增数字 const steps ref(new Map()) // 当前正在处理的stepId用于高亮 const activeStepId ref(null) // 计算属性按类型分组的步骤 const thoughts computed(() Array.from(steps.value.values()).filter(s s.step thought) ) const actions computed(() Array.from(steps.value.values()).filter(s s.step action) ) // 添加步骤方法供SSE事件处理器调用 function addStep(stepData) { const stepId stepData.id || Date.now().toString() // 后端应提供id steps.value.set(stepId, { id: stepId, ...stepData, timestamp: new Date().toISOString(), // 前端补充时间戳用于对比 verified: false, // 审计标记 expanded: false // 折叠状态 }) activeStepId.value stepId } // 审计操作标记为已验证 function verifyStep(stepId) { const step steps.value.get(stepId) if (step) step.verified true } // 清空所有步骤 function clearSteps() { steps.value.clear() activeStepId.value null } return { steps, activeStepId, thoughts, actions, addStep, verifyStep, clearSteps } })4.2 SSE连接管理用composable封装可复用的连接逻辑SSE连接不是一次性事件它需要自动重连、错误处理、连接状态反馈、取消机制。我们把它封装成useEventSourcecomposable// composables/useEventSource.js import { ref, onUnmounted } from vue export function useEventSource(url) { const eventSource ref(null) const status ref(idle) // connecting, open, error, closed const error ref(null) function connect() { if (eventSource.value) return status.value connecting eventSource.value new EventSource(url) eventSource.value.onopen () { status.value open error.value null } eventSource.value.onerror (e) { status.value error error.value e // 自动重连EventSource默认重连但这里可加退避策略 console.warn(SSE connection error, will retry...) } eventSource.value.addEventListener(init, (e) { console.log(Agent started:, JSON.parse(e.data)) }) eventSource.value.addEventListener(step, (e) { const step JSON.parse(e.data) // 触发全局事件让store处理 window.dispatchEvent(new CustomEvent(agent:step, { detail: step })) }) eventSource.value.addEventListener(done, (e) { console.log(Agent completed:, JSON.parse(e.data)) status.value closed }) } function disconnect() { if (eventSource.value) { eventSource.value.close() eventSource.value null status.value closed } } // 组件卸载时自动断开 onUnmounted(() { disconnect() }) return { eventSource, status, error, connect, disconnect } }在组件中使用!-- views/AgentView.vue -- script setup import { onMounted } from vue import { useAgentStore } from /stores/agent import { useEventSource } from /composables/useEventSource const agentStore useAgentStore() const { connect, status } useEventSource(/api/agent/stream?query encodeURIComponent(query)) // 监听全局step事件 window.addEventListener(agent:step, (e) { agentStore.addStep(e.detail) }) onMounted(() { connect() }) /script template div classagent-container div classstatus-bar span :class{ online: status open, offline: status ! open } {{ status open ? ✅ Connected : ⚠️ Connecting... }} /span /div div classsteps-list StepItem v-forstep of agentStore.steps.values() :keystep.id :stepstep / /div /div /template4.3 “可审核”UI的四个关键交互设计可审核性不是加个“导出JSON”按钮。它体现在交互细节里步骤折叠/展开Observation内容常很长如API返回的完整JSON默认折叠点击展开。StepItem组件里template div classstep-item :class{ active: step.id activeStepId } div classstep-header clickstep.expanded !step.expanded span classstep-type{{ step.step }}/span span classstep-timestamp{{ formatTime(step.timestamp) }}/span span classexpand-icon{{ step.expanded ? ▲ : ▼ }}/span /div div v-ifstep.expanded classstep-content pre{{ JSON.stringify(step, null, 2) }}/pre /div /div /template人工验证标记每个步骤旁有✅按钮点击后step.verified true背景变浅绿色。审计员可快速标记“这步思考合理”。时间轴视图用Timeline组件按timestamp排序直观展示Agent耗时分布比如thought占200msaction占3200msobservation占150ms。导出为审计报告点击“Export Audit Report”生成Markdown文件包含## ReAct Agent Audit Report - **Query**: 北京天气 - **Start Time**: 2024-05-20T14:23:05.123Z - **End Time**: 2024-05-20T14:23:07.456Z - **Total Steps**: 12 - **Verified Steps**: 8/12 ### Step Details 1. thought: 需要获取北京当前天气... - Verified: ✅ - Duration: 123ms 2. action: weather_api({city: 北京}) - Verified: ✅ - Duration: 3210ms ...这才是“可审核”的实质它把被动的日志阅读变成主动的、结构化的、可协作的审计过程。5. FastAPI后端ReAct Agent的流式执行引擎与SSE协议适配器FastAPI在这里的角色远不止“提供API”。它是ReAct Agent的执行沙箱、SSE协议的翻译器、以及审计元数据的注入点。很多教程只教app.get返回JSON但要支撑可审核Agent后端需深度介入执行流。5.1 ReAct Agent执行流的重构从同步到异步生成器标准ReAct实现常是同步函数# ❌ 同步实现无法流式推送 def run_react_agent(query: str) - dict: thought generate_thought(query) action parse_action(thought) observation execute_tool(action) answer generate_answer(observation) return {answer: answer, steps: [thought, action, observation]}这导致前端必须等全部完成才收到结果失去“可审核”基础。正确做法是用异步生成器每步完成后yield# fastapi_backend/agent/core.py import asyncio from typing import AsyncGenerator, Dict, Any async def run_react_agent_stream(query: str) - AsyncGenerator[Dict[str, Any], None]: Stream ReAct steps as they happen. Yields dict with keys: step, content, tool, params, observation, etc. # Step 1: Generate initial thought thought_content await llm_generate(fThink step by step to answer: {query}) yield { step: thought, content: thought_content, timestamp: asyncio.get_event_loop().time() } # Step 2: Parse action action_data await parse_action(thought_content) yield { step: action, tool: action_data[tool], params: action_data[params], timestamp: asyncio.get_event_loop().time() } # Step 3: Execute tool (with timeout error handling) try: observation await execute_tool(action_data[tool], action_data[params]) yield { step: observation, content: observation, timestamp: asyncio.get_event_loop().time() } except Exception as e: yield { step: error, content: fTool execution failed: {str(e)}, timestamp: asyncio.get_event_loop().time() } return # 终止流 # Step 4: Generate final answer answer await llm_generate(fGiven observation: {observation}, answer: {query}) yield { step: answer, content: answer, timestamp: asyncio.get_event_loop().time() }5.2 SSE端点如何让FastAPI正确流式响应FastAPI的Response支持流式但需注意几个坑必须设置media_typetext/event-stream否则浏览器不识别必须禁用Gzip压缩SSE不兼容压缩流必须处理客户端断开避免后端继续计算浪费资源必须添加Cache-Control: no-cache防止CDN缓存SSE流。完整端点实现# fastapi_backend/main.py from fastapi import Response, Request, HTTPException from fastapi.responses import StreamingResponse import asyncio import json from typing import AsyncGenerator app.get(/api/agent/stream) async def stream_agent_response( request: Request, query: str, timeout: int 300 # 最大执行时间单位秒 ): Stream ReAct Agent steps via SSE. Handles client disconnect gracefully. # 1. 创建生成器 agent_stream run_react_agent_stream(query) # 2. 定义流式生成器包装agent_stream处理中断 async def event_generator(): try: # 发送初始化事件 yield fevent: init\ndata: {json.dumps({query: query, started_at: asyncio.get_event_loop().time()})}\n\n # 3. 逐个yield agent步骤 async for step in agent_stream: # 检查客户端是否还连接 if await request.is_disconnected(): print(Client disconnected, stopping stream) break # 添加唯一ID用于SSE重连 step_id str(int(asyncio.get_event_loop().time() * 1000000)) yield fid: {step_id}\nevent: step\ndata: {json.dumps(step)}\n\n # 强制flush确保立即发送避免缓冲 await asyncio.sleep(0.001) # 4. 发送完成事件 yield fevent: done\ndata: {json.dumps({completed_at: asyncio.get_event_loop().time()})}\n\n except Exception as e: # 任何错误都转为error事件 error_step { step: error, content: fAgent execution error: {str(e)}, timestamp: asyncio.get_event_loop().time() } yield fevent: error\ndata: {json.dumps(error_step)}\n\n # 5. 返回StreamingResponse return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # Nginx专用禁用缓冲 } )5.3 审计增强在每步中注入可追溯元数据可审核性要求每步都带足够上下文。我们在run_react_agent_stream中注入trace_id全链路追踪ID关联CLI、后端、工具调用model_used实际调用的大模型名称gpt-4, claude-3等tool_duration_ms工具执行耗时毫秒llm_prompt_tokens本次LLM调用的输入token数。示例注入# 在yield前 import uuid from datetime import datetime trace_id str(uuid.uuid4()) start_time datetime.utcnow().isoformat() # 在yield thought时 yield { step: thought, content: thought_content, trace_id: trace_id, model_used: gpt-4-turbo, prompt_tokens: 123, timestamp: start_time } # 在yield observation时 end_time datetime.utcnow().isoformat() duration_ms (datetime.fromisoformat(end_time) - datetime.fromisoformat(start_time)).total_seconds() * 1000 yield { step: observation, content: observation, trace_id: trace_id, tool: weather_api, tool_duration_ms: round(duration_ms, 2), timestamp: end_time }这些字段在Vue前端可直接显示也可导出到审计报告形成完整证据链。6. 全链路联调与典型故障排查从“Hello World”到生产就绪写完代码只是开始。真正的挑战在联调阶段。根据我处理过的37个类似项目以下故障出现频率最高且都有确定性解法。6.1 故障1Vue前端收不到任何SSE事件Network面板显示pending现象浏览器DevTools Network标签页里/api/agent/stream请求状态一直是pending没有数据。根因分析后端未正确设置Content-Type: text/event-stream后端未发送任何data:消息比如忘了yield或生成器为空反向代理Nginx未配置SSE支持。排查链路先curl测试curl -N http://localhost:8000/api/agent/stream?querytest。如果curl也pending问题在后端或代理。检查FastAPI响应头在curl命令后加-I看是否含Content-Type: text/event-stream。若无检查FastAPI代码中media_type参数。检查后端日志在event_generator()里加print(Sending init event)确认后端是否执行到yield。Nginx配置检查location /api/agent/stream { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; proxy_cache_bypass $http_upgrade; # 关键禁用缓冲 proxy_buffering off; proxy_buffer_size 4k; proxy_buffers 8 4k; # 关键长连接超时 proxy_read_timeout 300; }6.2 故障2SSE连接频繁断开报错stream disconnected before completion: idle timeout waiting for sse现象前端EventSource反复断开重连控制台刷屏error事件。根因分析Uvicorn的--timeout-keep-alive值太小默认5秒后端Agent某步执行超时未及时发送心跳云服务商AWS ALB, Cloudflare的空闲超时设置过短常为60秒。解决方案Uvicorn启动参数uvicorn main:app --timeout-keep-alive 3005分钟后端主动心跳在event_generator()循环中每30秒yield data: {}\n\n空消息云服务配置AWS ALB健康检查路径设为/health空闲超时调至300秒Cloudflare在Rules Transform Rules中添加规则将/api/agent/stream的超时设为300秒。6.3 故障3CLI报错unable to locate the codex cli binary or required runtime components现象用户下载cod