1. 这不是又一个“聊天界面”而是一套可审计的智能体执行流水线上周五下午三点我盯着终端里一行行滚动的curl -N http://localhost:8000/agent/stream输出发了三分钟呆——不是因为卡顿而是因为终于看到{step:execute,tool:search_web,query:2024年Q3全球AI芯片出货量统计}这样的结构化日志像工厂流水线上的工单编号一样稳稳地从 FastAPI 后端流进 Vue 前端再被实时渲染成带时间戳、带步骤类型、带工具调用参数的卡片。那一刻我才真正意识到我们做的不是“让大模型说话”而是给 AI 的思考过程装上仪表盘和黑匣子。这个项目标题里的每个词都不是装饰CLI 是起点它代表命令行下可复现、可脚本化的最小验证单元浏览器是终点但不是炫技的 UI而是面向产品、运营、法务甚至审计人员的可视化操作台FastAPI 是承重墙它不只提供/chat接口更要承载状态管理、流控、日志注入、权限校验四层责任SSE 是血管它比 WebSocket 更轻量、比轮询更实时专为“单向、长时、事件驱动”的推理流设计Vue 3 是神经末梢用 Composition API 精准响应每一条data: {type:thought,content:需要验证数据来源可靠性...}而ReAct Agent 是灵魂——它不是把 prompt 拼得更长而是用明确的Thought/Action/Observation/Answer四段式结构把“AI 怎么想的”变成可拆解、可回溯、可人工干预的原子操作。你可能刚在某篇教程里跑通过fastapi vue的 hello world也可能用过codex cli或zcode cli调用本地模型。但那些 demo 里response.text是一团不可分割的字符串console.log(data)只能看到最终答案。而本项目要解决的是真实业务场景里的三个硬需求第一当客户质疑“为什么推荐这款芯片”你能立刻拉出第 7 步的search_web调用记录和返回的原始网页快照第二当线上 agent 卡在Observation阶段超过 15 秒运维能通过 FastAPI 的/health接口 Prometheus 指标精准定位是模型响应超时还是网络抖动第三当合规部门要求“所有工具调用必须经审批”你在action_router.py里加一行if action send_email: raise PermissionError(Email requires manual approval)就能生效——这些能力全系于 CLI 到浏览器这条链路的每一环是否真正“可审计”。提示本项目不依赖任何闭源 SDK 或云服务。所有代码基于 Python 3.11、FastAPI 0.111、Vue 3.4Composition API Pinia、Vite 构建。核心逻辑全部开源你可以把它嵌入现有企业内网系统也可以作为独立服务部署在私有 Kubernetes 集群中。2. CLI 层用纯 Python 实现 ReAct 循环拒绝黑盒封装很多初学者一上来就冲着langchain或llamaindex的AgentExecutor去结果调试时发现agent.run()抛出异常连哪一步Thought出错都看不到。本项目的第一步就是亲手用 200 行纯 Python 写一个最小可运行的 ReAct 循环——它不漂亮但每一行都在你眼皮底下。2.1 ReAct 的本质不是 Prompt 工程而是状态机驱动ReAct 的核心不是“让模型学会思考”而是定义一套人类可读、机器可执行的状态转移规则。我们把整个循环抽象为四个状态THOUGHT: 模型输出一段自然语言描述当前推理路径如“需要查证该芯片的功耗数据是否符合客户要求”ACTION: 模型按固定格式输出工具调用指令如Action: search_web\nAction Input: {query:NVIDIA H100 250W TDP official spec}OBSERVATION: 工具执行后返回的原始结果如htmltitleNVIDIA H100 Data Sheet/title...ANSWER: 模型综合所有 Observation 后给出最终回答如“H100 的典型功耗为 250W符合客户 300W 以内要求”关键在于状态切换必须由代码显式控制而非依赖模型“自觉”输出特定 token。我们用一个while True循环 state变量实现# cli/agent_core.py from typing import Dict, Any, Optional import json import re class ReActAgent: def __init__(self, llm_client): self.llm_client llm_client # 支持 openai.Completion 或 ollama.chat self.history [] # 存储完整的 step-by-step 日志 def run(self, user_query: str) - str: self.history.clear() self.history.append({role: user, content: user_query}) max_steps 8 for step in range(max_steps): # 1. 生成 Thought Action prompt self._build_prompt() response self.llm_client.generate(prompt) # 2. 解析模型输出严格匹配正则 thought_match re.search(rThought:\s*(.*?)(?:\n|$), response) action_match re.search(rAction:\s*(\w)\nAction Input:\s*(\{.*?\}), response, re.DOTALL) if not thought_match or not action_match: # 解析失败强制进入 ANSWER 状态 self.history.append({role: assistant, content: fFailed to parse step {step}}) break thought thought_match.group(1).strip() action_name action_match.group(1).strip() try: action_input json.loads(action_match.group(2)) except json.JSONDecodeError: action_input {raw: action_match.group(2)} # 3. 记录 Thought Action 到 history self.history.append({ step: step, type: thought, content: thought, timestamp: time.time() }) self.history.append({ step: step, type: action, tool: action_name, input: action_input, timestamp: time.time() }) # 4. 执行工具并记录 Observation observation self._execute_tool(action_name, action_input) self.history.append({ step: step, type: observation, tool: action_name, output: observation[:500] ... if len(observation) 500 else observation, timestamp: time.time() }) # 5. 如果是 final answer跳出循环 if action_name finish: return observation return Max steps exceeded注意_execute_tool方法是真正的业务胶水。本项目预置了search_web调用 duckduckgo-search、get_weather调用 OpenWeatherMap API、read_file读取本地 JSON三个工具。每个工具都必须返回结构化字典如{status: success, data: {...}}而非原始 HTML 字符串。这是后续审计的关键——Observation字段必须能被前端直接解析而不是扔给用户一堆div标签。2.2 CLI 的价值可复现、可压测、可集成到 CI/CD写完ReActAgent类我们立刻用argparse包装成 CLI 工具# cli/main.py import argparse from agent_core import ReActAgent from llm_clients import OllamaClient # 或 OpenAIClient def main(): parser argparse.ArgumentParser(descriptionRun ReAct Agent from CLI) parser.add_argument(--query, typestr, requiredTrue, helpUser query) parser.add_argument(--model, typestr, defaultllama3, helpLLM model name) parser.add_argument(--max-steps, typeint, default8, helpMax ReAct steps) args parser.parse_args() client OllamaClient(modelargs.model) agent ReActAgent(client) result agent.run(args.query) # 关键输出完整 history 为 JSONL供后续分析 for log in agent.history: print(json.dumps(log, ensure_asciiFalse)) if __name__ __main__: main()执行python cli/main.py --query 对比 RTX 4090 和 H100 在 AI 训练场景的性价比你会得到 10 行 JSONL 输出每行是一个带type、step、timestamp的审计事件。这带来三个实操优势调试效率翻倍当某次运行卡住你不用重启整个 Web 服务只需grep type:action output.jsonl | tail -3查看最后三次工具调用立刻判断是search_web返回空结果还是get_weatherAPI 密钥失效压测脚本直连用ab或wrk对 FastAPI 接口压测时后端实际调用的就是这个 CLI 类。你可以在pytest中写test_agent_step_by_step.py用mock.patch替换OllamaClient100% 覆盖所有Thought/Action/Observation分支CI/CD 自动化审计在 GitHub Actions 中每次 PR 提交后自动运行python cli/main.py --query 测试用例将history输出存为 artifact。如果某次type字段出现error或timeoutPipeline 直接失败——这才是真正的“可审计”。实操心得我在第一次部署时发现search_web工具在服务器上因 DNS 解析超时导致整个 agent 卡死。解决方案不是加try/except而是在 CLI 层增加-t/--timeout参数并在_execute_tool中统一用requests.get(url, timeoutargs.timeout)。这样审计日志里会明确记录type:error,message:Timeout on search_web而不是让前端显示一片空白。3. FastAPI 层不止是 API Server更是审计日志的中央枢纽很多 FastAPI 教程教你写app.post(/chat)然后return {response: llm.generate(...)}。但在可审计 agent 场景下FastAPI 必须承担四项额外职责流式响应封装、请求上下文注入、结构化日志落库、跨域与安全加固。我们逐项拆解。3.1 SSE 流的本质HTTP Chunked Transfer Encoding 的优雅封装SSEServer-Sent Events不是魔法它是 HTTP/1.1 的Transfer-Encoding: chunked特性 服务端主动推送的组合。FastAPI 用StreamingResponse实现它但关键在于如何把 ReAct 的离散步骤转换成符合 SSE 规范的连续数据流。标准 SSE 格式要求每条消息以data:开头后跟 JSON 字符串消息间用双换行分隔可选event:定义事件类型如event: thought可选id:用于客户端断线重连我们的stream_agent接口这样实现# api/main.py from fastapi import FastAPI, Request, Depends from fastapi.responses import StreamingResponse from starlette.concurrency import iterate_in_threadpool import json import time from cli.agent_core import ReActAgent from llm_clients import OllamaClient app FastAPI() app.post(/agent/stream) async def stream_agent(request: Request): # 1. 解析请求体获取 query 和 session_id body await request.json() user_query body.get(query, ) session_id body.get(session_id, fsess_{int(time.time())}) # 2. 初始化 agent注意这里不能 new ReActAgent()需注入 context agent ReActAgent(OllamaClient(modelllama3)) # 3. 定义生成器函数yield 每个 step async def event_generator(): try: # 发送初始化事件 yield fevent: init\ndata: {json.dumps({session_id: session_id, timestamp: time.time()})}\n\n # 执行 agent.run()但捕获每一步的 history # 注意这里不能直接调用 agent.run()因为它返回最终字符串 # 我们需要重写 run() 为 generator for step_log in agent.run_stream(user_query): # 新增方法 # 4. 格式化为 SSE 消息 yield fevent: {step_log[type]}\ndata: {json.dumps(step_log, ensure_asciiFalse)}\n\n # 5. 强制 flush避免 Nginx 缓存 await asyncio.sleep(0.01) except Exception as e: yield fevent: error\ndata: {json.dumps({error: str(e), timestamp: time.time()})}\n\n return StreamingResponse( event_generator(), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no # 关键禁用 Nginx 缓存 } )agent.run_stream()是对原run()方法的重构它不再返回字符串而是yield每个step_log# cli/agent_core.py def run_stream(self, user_query: str): self.history.clear() self.history.append({role: user, content: user_query}) for step in range(self.max_steps): # ... [同前] ... # 在每次 self.history.append() 后yield 当前 log yield self.history[-1] # 最新一条日志 # ... [继续] ...关键细节X-Accel-Buffering: no头是生产环境的救命稻草。没有它Nginx 会默认缓存 8KB 数据才推送给前端导致 SSE 消息延迟数秒甚至超时。stream disconnected before completion: idle timeout waiting for sse这个热词错误90% 源于此。我们在nginx.conf中还额外配置了proxy_buffering off; proxy_cache off;双重保险。3.2 审计日志的落地不只是 print而是结构化存储CLI 层的print(json.dumps(log))只适合开发。生产环境必须把每条step_log存入数据库且满足审计要求不可篡改、带时间戳、关联 session_id、支持 SQL 查询。我们选用 SQLite轻量 SQLAlchemy Core非 ORM避免性能损耗# api/db.py from sqlalchemy import create_engine, text from sqlalchemy.pool import StaticPool # 使用 StaticPool 避免多线程连接问题 engine create_engine( sqlite:///./audit.db, connect_args{check_same_thread: False}, poolclassStaticPool ) # 初始化表 with engine.connect() as conn: conn.execute(text( CREATE TABLE IF NOT EXISTS audit_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, step INTEGER NOT NULL, type TEXT NOT NULL CHECK(type IN (thought,action,observation,answer,error)), tool TEXT, content TEXT, timestamp REAL NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) )) conn.commit()在stream_agent的event_generator中每yield一条日志就同步插入数据库# api/main.py from api.db import engine async def event_generator(): # ... [初始化] ... for step_log in agent.run_stream(user_query): # 插入审计日志同步操作确保顺序 with engine.connect() as conn: conn.execute(text( INSERT INTO audit_logs (session_id, step, type, tool, content, timestamp) VALUES (:session_id, :step, :type, :tool, :content, :timestamp) ), { session_id: session_id, step: step_log.get(step, 0), type: step_log[type], tool: step_log.get(tool, ), content: json.dumps(step_log, ensure_asciiFalse)[:1000], # 防止超长 timestamp: step_log[timestamp] }) conn.commit() yield fevent: {step_log[type]}\ndata: {json.dumps(step_log, ensure_asciiFalse)}\n\n实操心得不要用异步数据库驱动如 asyncpg处理审计日志。SSE 流要求事件严格按序而异步 I/O 可能导致日志入库顺序与流输出顺序不一致。我们实测过用threading.Lock()包裹conn.execute()比异步方案更稳定。另外content字段限制 1000 字符是因为 SQLite 的TEXT类型虽无硬限制但过长字段会显著拖慢SELECT * FROM audit_logs WHERE session_idxxx查询速度。3.3 CORS 与安全加固FastAPI 不是裸奔的玩具fastapi cors是高频搜索词说明很多人栽在跨域上。但真正的风险不在allow_origins[*]而在于未校验 session_id、未限制请求频率、未过滤恶意 query。我们添加三层防护Session ID 校验前端必须在请求体中传session_id后端用uuid.uuid4()生成并返回后续请求必须携带。防止恶意脚本批量调用速率限制用slowapi库限制/agent/stream每 IP 每分钟 5 次Query 过滤对user_query做基础清洗移除\x00-\x08\x0b\x0c\x0e-\x1f等控制字符防止注入攻击。# api/main.py from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.post(/agent/stream) limiter.limit(5/minute) async def stream_agent(request: Request): body await request.json() user_query body.get(query, ) session_id body.get(session_id, ) # 1. Session ID 校验 if not session_id or not re.match(r^sess_\d$, session_id): raise HTTPException(status_code400, detailInvalid session_id) # 2. Query 清洗 clean_query re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f], , user_query) if len(clean_query) 3 or len(clean_query) 500: raise HTTPException(status_code400, detailQuery too short or too long) # ... [后续逻辑] ...注意fastapi cors错误常源于 Nginx 配置。我们nginx.conf中明确设置location /api/ { proxy_pass http://fastapi_backend; proxy_set_header Origin $scheme://$host; add_header Access-Control-Allow-Origin $scheme://$host; add_header Access-Control-Allow-Methods GET, POST, OPTIONS; add_header Access-Control-Allow-Headers Content-Type, Authorization; }这样前端fetch(http://your-domain.com/api/agent/stream)才能正确拿到 CORS 头。4. Vue 3 层用 Composition API 构建可追溯的 UI 状态机Vue 3 的 Composition API 不是语法糖它是构建复杂状态流的刚需。本项目的前端不是“展示数据”而是精确映射 ReAct 的四个状态并允许用户在任意步骤暂停、重试、导出。4.1 Pinia Store定义 agent 的单一事实源我们创建stores/agent.ts用defineStore管理 agent 的完整生命周期// stores/agent.ts import { defineStore } from pinia import { ref, computed } from vue export interface StepLog { step: number type: thought | action | observation | answer | error tool?: string input?: Recordstring, any output?: string content?: string timestamp: number } export const useAgentStore defineStore(agent, () { const sessionId refstring() const isStreaming refboolean(false) const logs refStepLog[]([]) const currentStep refnumber(0) const status refidle | thinking | executing | observing | answering | error(idle) // 计算属性按 step 分组的日志 const groupedLogs computed(() { return logs.value.reduce((acc, log) { if (!acc[log.step]) acc[log.step] [] acc[log.step].push(log) return acc }, {} as Recordnumber, StepLog[]) }) // 计算属性当前 step 的最新状态 const currentStatus computed(() { const lastLog logs.value[logs.value.length - 1] if (!lastLog) return idle switch (lastLog.type) { case thought: return thinking case action: return executing case observation: return observing case answer: return answering case error: return error default: return idle } }) // Action启动 stream const startStream async (query: string) { isStreaming.value true sessionId.value sess_${Date.now()} logs.value [] try { const response await fetch(/api/agent/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ query, session_id: sessionId.value }) }) if (!response.ok) throw new Error(HTTP ${response.status}) const reader response.body?.getReader() if (!reader) throw new Error(No reader) while (true) { const { done, value } await reader.read() if (done) break const decoder new TextDecoder() const text decoder.decode(value) const lines text.split(\n) for (const line of lines) { if (line.startsWith(data:)) { try { const data JSON.parse(line.substring(5).trim()) logs.value.push(data) } catch (e) { console.warn(Invalid SSE data:, line) } } } } } catch (e) { logs.value.push({ step: logs.value.length, type: error, content: Stream failed: ${(e as Error).message}, timestamp: Date.now() }) } finally { isStreaming.value false } } // Action导出当前 session 日志 const exportLogs () { const blob new Blob([JSON.stringify(logs.value, null, 2)], { type: application/json }) const url URL.createObjectURL(blob) const a document.createElement(a) a.href url a.download agent_session_${sessionId.value}.json a.click() URL.revokeObjectURL(url) } return { sessionId, isStreaming, logs, currentStep, status, groupedLogs, currentStatus, startStream, exportLogs } })关键设计groupedLogs计算属性把logs数组按step分组这样模板里可以用div v-for(stepLogs, step) in groupedLogs渲染每一步的卡片。每个卡片包含thought、action、observation三条日志形成完整的“推理单元”。4.2 组件化渲染每个 step 是一个可交互的审计单元AgentView.vue组件的核心是v-for渲染groupedLogs!-- components/AgentView.vue -- template div classagent-container !-- 输入区 -- div classinput-section input v-modelquery keyup.enterstartStream placeholder输入你的问题... :disabledisStreaming / button clickstartStream :disabledisStreaming {{ isStreaming ? 运行中... : 开始推理 }} /button /div !-- 日志流 -- div classlogs-section div v-for(stepLogs, step) in groupedLogs :keystep classstep-card div classstep-header span classstep-numberStep {{ step }}/span span classstep-status{{ getStatusText(stepLogs) }}/span /div !-- Thought -- div v-ifgetLogByType(stepLogs, thought) classlog-item thought strongThought:/strong {{ getLogByType(stepLogs, thought)!.content }} /div !-- Action -- div v-ifgetLogByType(stepLogs, action) classlog-item action strongAction:/strong {{ getLogByType(stepLogs, action)!.tool }} span v-ifgetLogByType(stepLogs, action)!.input ({{ JSON.stringify(getLogByType(stepLogs, action)!.input, null, 2) }}) /span /div !-- Observation -- div v-ifgetLogByType(stepLogs, observation) classlog-item observation strongObservation:/strong pre{{ getLogByType(stepLogs, observation)!.output }}/pre /div !-- Answer 或 Error -- div v-ifgetLogByType(stepLogs, answer) classlog-item answer strongAnswer:/strong {{ getLogByType(stepLogs, answer)!.content }} /div div v-ifgetLogByType(stepLogs, error) classlog-item error strongError:/strong {{ getLogByType(stepLogs, error)!.content }} /div /div /div !-- 控制区 -- div classcontrol-section v-iflogs.length 0 button clickexportLogs导出本次会话日志/button button clickclearLogs清空日志/button /div /div /template script setup langts import { ref, computed } from vue import { useAgentStore } from /stores/agent const store useAgentStore() const query ref() const startStream () { if (!query.value.trim()) return store.startStream(query.value) query.value } const clearLogs () { store.logs [] store.sessionId } // 辅助函数根据 type 获取日志 const getLogByType (logs: StepLog[], type: string) { return logs.find(log log.type type) } const getStatusText (logs: StepLog[]) { const last logs[logs.length - 1] if (last.type answer) return 完成 if (last.type error) return 错误 return last.type thought ? 思考中 : last.type action ? 执行中 : 观察中 } /script实操心得pre标签渲染Observation是刻意为之。很多教程用v-html但这有 XSS 风险。我们要求所有Observation输出必须是纯文本或 JSON前端不做任何 HTML 解析。如果工具返回 HTML_execute_tool方法必须先用BeautifulSoup提取文本再存入output字段。这样审计日志里存的是干净文本前端展示也绝对安全。4.3 Vue 3 Snippets提升开发效率的实战技巧vue 3 snippets是高频搜索词说明开发者渴望开箱即用的代码片段。我们整理了三个高频场景的 snippetSSE 连接重试当网络中断前端自动重连// utils/sse-reconnect.ts export function createSSEWithRetry(url: string, onMessage: (data: any) void) { let eventSource: EventSource | null null const connect () { eventSource new EventSource(url) eventSource.onmessage (e) onMessage(JSON.parse(e.data)) eventSource.onerror () { console.warn(SSE connection lost, retrying in 3s...) setTimeout(connect, 3000) } } connect() return () eventSource?.close() }日志高亮用不同颜色区分Thought/Action/Observation/* styles/agent.css */ .log-item.thought { background-color: #e6f7ff; border-left: 4px solid #1890ff; } .log-item.action { background-color: #fff0f6; border-left: 4px solid #eb2f96; } .log-item.observation { background-color: #f6ffed; border-left: 4px solid #52c418; } .log-item.answer { background-color: #f0f9ff; border-left: 4px solid #1890ff; font-weight: bold; } .log-item.error { background-color: #fff2f0; border-left: 4px solid #f5222d; }响应式布局适配移动端用media优化小屏体验media (max-width: 768px) { .step-card { padding: 12px; } .step-header { flex-direction: column; } .log-item pre { white-space: pre-wrap; word-break: break-word; } }注意vs code gemini cli companion或claude code cli这类工具本质是把 LLM 的代码补全能力接入 IDE。但本项目强调前端逻辑必须手写不能依赖 AI 生成。因为审计要求“代码可知、行为可溯”。我们用vue-tsc --noEmit做类型检查用vitest写单元测试确保getLogByType等辅助函数 100% 覆盖。5. 全链路联调与生产级避坑指南当 CLI、FastAPI、Vue 三端各自跑通真正的挑战才开始如何让它们在真实网络环境下稳定协同这里没有银弹只有踩过的坑和验证过的方案。5.1 “Stream disconnected before completion” 的七种根因与修复这个错误是 SSE 最常见的报错但原因千差万别。我们按发生位置分类位置根因修复方案验证命令客户端浏览器主动关闭连接用户切页前端监听visibilitychange事件页面隐藏时暂停 streamdocument.addEventListener(visibilitychange, () { if (document.hidden) controller.abort() })Nginxproxy_read_timeout默认 60s在nginx.conf中设proxy_read_timeout 300;curl -N http://localhost/api/agent/stream观察是否 60s 后断开FastAPIStreamingResponse未及时 flush加await asyncio.sleep(0.01)强制 flush在event_generator中添加print(flushed)日志LLM ClientOllamaClient请求超时在llm_clients.py中设requests.post(..., timeout120)time python cli/main.py --query long query数据库INSERT操作阻塞流改为异步写入用asyncio.to_thread包裹ab -n 100 -c 10 http://localhost/api/agent/stream防火墙云服务商拦截长连接开放 TCP keepalive设net.ipv4.tcp_keepalive_time600ss -tn前端EventSource未处理error事件添加eventSource.onerror () { /* 重试逻辑 */ }手动断开网络观察前端是否重连实测案例某次上线后AWS ALB 报错stream disconnected before completion: idle timeout waiting for sse。排查发现 ALB 的空闲超时默认 60 秒而我们的search_web工具在高峰时段响应达 90 秒。解决方案不是改工具而是在 ALB 控制台将Idle timeout改为 300 秒并在 FastAPI 中加app.middleware(http)记录每个请求的time.time()对比日志确认是 ALB 断连。5.2 从 CLI 到浏览器的端到端测试脚本自动化测试不是可选项而是审计合规的基石。我们用playwright写了一个端到端测试# tests/e2e_test.py from playwright.sync_api import sync_playwright import json import time def test_react_agent_flow(): with sync_playwright() as p: browser p.chromium.launch