diff --git a/backend/ai_agent.py b/backend/ai_agent.py index 639fdfb..9107004 100644 --- a/backend/ai_agent.py +++ b/backend/ai_agent.py @@ -1,4 +1,5 @@ -"""AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果""" +"""AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果 (支持流式输出)""" +import json import httpx from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL @@ -13,68 +14,111 @@ SYSTEM_PROMPT = """你是一位资深的 Kubernetes 运维专家和 SRE 工程 输出格式：使用清晰的 Markdown，包含标题、列表、代码块。语言使用中文。""" +_HEADERS = { + "Authorization": f"Bearer {AI_API_KEY}", + "Content-Type": "application/json", +} -async def analyze_with_ai(report: str, question: str = "") -> dict: - """调用 AI 分析诊断报告""" + +async def _stream_chat(messages: list[dict]): + """通用流式请求 - yield 文本 chunk""" + # connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死 + timeout = httpx.Timeout(90, connect=10) + async with httpx.AsyncClient(timeout=timeout) as client: + async with client.stream( + "POST", + f"{AI_API_BASE}/chat/completions", + headers=_HEADERS, + json={ + "model": AI_MODEL, + "messages": messages, + "temperature": 0.3, + "max_tokens": 4096, + "stream": True, + }, + ) as resp: + # 非 2xx 直接抛异常 (由上层转成 error 事件) + resp.raise_for_status() + + # 有些服务认证失败时返回 200 + JSON 错误体 (非 SSE)，需要识别 + ctype = resp.headers.get("content-type", "") + if "text/event-stream" not in ctype and "application/json" in ctype: + body = await resp.aread() + try: + err = json.loads(body.decode()) + raise httpx.HTTPStatusError( + f"AI 服务返回错误: {err.get('error', body.decode()[:300])}", + request=resp.request, + response=resp, + ) + except (json.JSONDecodeError, UnicodeDecodeError): + raise httpx.HTTPStatusError( + f"AI 服务返回非流式响应: {body.decode()[:300]}", + request=resp.request, + response=resp, + ) + + async for line in resp.aiter_lines(): + if not line.startswith("data: "): + continue + data_str = line[6:].strip() + if data_str == "[DONE]": + break + try: + chunk = json.loads(data_str) + delta = chunk["choices"][0].get("delta", {}) + content = delta.get("content", "") + if content: + yield content + except (json.JSONDecodeError, KeyError, IndexError): + continue + + +async def analyze_with_ai_stream(report: str, question: str = ""): + """流式 AI 分析诊断报告 - yield 文本 chunk""" user_msg = f"以下是 K8S 集群诊断报告，请进行详细分析：\n\n{report}" if question: user_msg += f"\n\n用户额外问题：{question}" - + + messages = [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": user_msg}, + ] + async for chunk in _stream_chat(messages): + yield chunk + + +async def chat_with_ai_stream(messages: list[dict], context: str = ""): + """流式多轮对话 - yield 文本 chunk""" + msgs = [{"role": "system", "content": SYSTEM_PROMPT}] + if context: + msgs.append({"role": "system", "content": f"当前诊断上下文：\n{context}"}) + msgs.extend(messages) + async for chunk in _stream_chat(msgs): + yield chunk + + +async def analyze_with_ai(report: str, question: str = "") -> dict: + """非流式 AI 分析 (兼容旧接口)""" try: - async with httpx.AsyncClient(timeout=120) as client: - resp = await client.post( - f"{AI_API_BASE}/chat/completions", - headers={ - "Authorization": f"Bearer {AI_API_KEY}", - "Content-Type": "application/json", - }, - json={ - "model": AI_MODEL, - "messages": [ - {"role": "system", "content": SYSTEM_PROMPT}, - {"role": "user", "content": user_msg}, - ], - "temperature": 0.3, - "max_tokens": 4096, - }, - ) - resp.raise_for_status() - data = resp.json() - content = data["choices"][0]["message"]["content"] - return {"success": True, "analysis": content, "model": AI_MODEL} + content = "" + async for chunk in analyze_with_ai_stream(report, question): + content += chunk + return {"success": True, "analysis": content, "model": AI_MODEL} except httpx.HTTPStatusError as e: return {"success": False, "analysis": f"AI 接口返回错误: {e.response.status_code} - {e.response.text[:500]}", "model": AI_MODEL} except httpx.ConnectError: - return {"success": False, "analysis": f"无法连接 AI 服务 ({AI_API_BASE})，请检查配置", "model": AI_MODEL} + return {"success": False, "analysis": f"无法连接 AI 服务 ({AI_API_BASE})，请检查 .env 中的 AI_API_BASE 配置", "model": AI_MODEL} except Exception as e: return {"success": False, "analysis": f"AI 分析失败: {str(e)}", "model": AI_MODEL} async def chat_with_ai(messages: list[dict], context: str = "") -> dict: - """多轮对话 - 基于诊断上下文追问""" - msgs = [{"role": "system", "content": SYSTEM_PROMPT}] - if context: - msgs.append({"role": "system", "content": f"当前诊断上下文：\n{context}"}) - msgs.extend(messages) - + """非流式多轮对话 (兼容旧接口)""" try: - async with httpx.AsyncClient(timeout=120) as client: - resp = await client.post( - f"{AI_API_BASE}/chat/completions", - headers={ - "Authorization": f"Bearer {AI_API_KEY}", - "Content-Type": "application/json", - }, - json={ - "model": AI_MODEL, - "messages": msgs, - "temperature": 0.3, - "max_tokens": 4096, - }, - ) - resp.raise_for_status() - data = resp.json() - content = data["choices"][0]["message"]["content"] - return {"success": True, "reply": content} + content = "" + async for chunk in chat_with_ai_stream(messages, context): + content += chunk + return {"success": True, "reply": content} except Exception as e: return {"success": False, "reply": f"AI 对话失败: {str(e)}"} diff --git a/backend/main.py b/backend/main.py index 6264c6b..c06f3e0 100644 --- a/backend/main.py +++ b/backend/main.py @@ -2,6 +2,7 @@ import os import json import asyncio +import httpx from datetime import datetime from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware @@ -10,7 +11,7 @@ from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import Optional -from backend.config import PORT +from backend.config import PORT, AI_API_BASE, AI_MODEL from backend.diagnosis import ( run_full_diagnosis, build_diagnosis_report, ALL_CHECKS, check_connectivity, @@ -18,7 +19,10 @@ from backend.diagnosis import ( check_deployments, check_services, check_pvc, check_resource_usage, check_network_policies, ) -from backend.ai_agent import analyze_with_ai, chat_with_ai +from backend.ai_agent import ( + analyze_with_ai, chat_with_ai, + analyze_with_ai_stream, chat_with_ai_stream, +) app = FastAPI(title="K8S 智能诊断平台", version="1.0.0") @@ -196,7 +200,7 @@ async def get_latest(): @app.post("/api/analyze") async def analyze(req: AnalyzeRequest): - """AI 分析诊断结果""" + """AI 分析诊断结果 (非流式，兼容旧接口)""" global _last_diagnosis, _last_report if _last_diagnosis is None: _last_diagnosis = await run_full_diagnosis() @@ -205,13 +209,73 @@ async def analyze(req: AnalyzeRequest): return result +@app.get("/api/analyze/stream") +async def analyze_stream(question: str = ""): + """SSE 流式 AI 分析 - 逐字推送分析内容""" + global _last_diagnosis, _last_report + if _last_diagnosis is None: + _last_diagnosis = await run_full_diagnosis() + _last_report = build_diagnosis_report(_last_diagnosis) + + async def event_generator(): + process_steps = [ + ("report", "读取诊断报告", "已加载本次集群诊断上下文"), + ("risk", "识别关键风险", "正在梳理严重问题、警告和潜在影响"), + ("solution", "生成修复方案", "正在为高优先级问题生成可执行操作"), + ("summary", "整理分析结论", "正在组织根因、影响和预防建议"), + ] + + yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL}, ensure_ascii=False)}\n\n" + try: + for step, title, detail in process_steps: + yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'running'}, ensure_ascii=False)}\n\n" + await asyncio.sleep(0) + yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'done'}, ensure_ascii=False)}\n\n" + + async for chunk in analyze_with_ai_stream(_last_report, question=question): + yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n" + yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" + except httpx.HTTPStatusError as e: + err = f"AI 接口返回错误: {e.response.status_code} - {str(e)[:300]}" + yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n" + except httpx.ConnectError: + yield f"data: {json.dumps({'type': 'error', 'message': f'无法连接 AI 服务 ({AI_API_BASE})，请检查 .env 中的 AI_API_BASE 配置'}, ensure_ascii=False)}\n\n" + except Exception as e: + yield f"data: {json.dumps({'type': 'error', 'message': f'AI 分析失败: {str(e)}'}, ensure_ascii=False)}\n\n" + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + @app.post("/api/chat") async def chat(req: ChatRequest): - """AI 多轮对话 (基于诊断上下文追问)""" + """AI 多轮对话 (非流式，兼容旧接口)""" result = await chat_with_ai(req.messages, context=_last_report) return result +@app.post("/api/chat/stream") +async def chat_stream(req: ChatRequest): + """SSE 流式 AI 对话 - 逐字推送回复""" + async def event_generator(): + yield f"data: {json.dumps({'type': 'start'}, ensure_ascii=False)}\n\n" + try: + async for chunk in chat_with_ai_stream(req.messages, context=_last_report): + yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n" + yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" + except Exception as e: + yield f"data: {json.dumps({'type': 'error', 'message': f'AI 对话失败: {str(e)}'}, ensure_ascii=False)}\n\n" + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + @app.get("/api/report") async def get_report(): """获取诊断报告文本""" diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 7e4c5bb..2ed61fd 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -207,34 +207,72 @@ - - - -
- - - - - -
-
-
-
-
-
思考中...
-
-
-
- - 发送 -
- + + + + +
+
+ 分析过程 + {{ analysisCompleted }}/{{ analysisProcess.length }} +
+ +
+
+ + + + + +
+ {{ item.title }} + {{ item.detail }} +
+ 处理中 + 完成 + 等待 +
+
+
+ +
+ + AI 正在生成分析结论，请稍候... +
+
+ + + + + +
+
+
+
+
+
思考中...
+
+
+
+ + 发送 +
+ @@ -253,6 +291,7 @@ const analyzing = ref(false) const diagnosis = ref(null) const aiAnalysis = ref('') const aiModel = ref('') +const analysisProcess = ref([]) const chatMessages = ref([]) const chatInput = ref('') const chatLoading = ref(false) @@ -281,6 +320,12 @@ const warningEvents = computed(() => (diagnosis.value?.checks?.events?.events || const resourceNodes = computed(() => diagnosis.value?.checks?.resource_usage?.nodes || []) const renderedAnalysis = computed(() => aiAnalysis.value ? renderMd(aiAnalysis.value) : '') +const analysisCompleted = computed(() => analysisProcess.value.filter(item => item.status === 'done').length) +const analysisProgressPercent = computed(() => { + if (!analysisProcess.value.length) return 0 + return Math.round((analysisCompleted.value / analysisProcess.value.length) * 100) +}) +const analysisProgressStatus = computed(() => analysisProgressPercent.value === 100 ? 'success' : '') function renderMd(text) { return marked.parse(text || '', { breaks: true }) @@ -357,17 +402,60 @@ async function runDiagnosis() { async function runAnalysis() { analyzing.value = true + aiAnalysis.value = '' + aiModel.value = '' + analysisProcess.value = [] try { - const resp = await fetch(`${API}/analyze`, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({}), - }) - const data = await resp.json() - aiAnalysis.value = data.analysis || '' - aiModel.value = data.model || '' - if (!data.success) ElMessage.warning('AI 分析异常') + const resp = await fetch(`${API}/analyze/stream`) + const reader = resp.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + + while (true) { + const { done, value } = await reader.read() + if (done) break + + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() || '' + + for (const line of lines) { + if (!line.startsWith('data: ')) continue + const jsonStr = line.slice(6) + if (!jsonStr) continue + try { + const msg = JSON.parse(jsonStr) + if (msg.type === 'start') { + aiModel.value = msg.model || '' + } else if (msg.type === 'process') { + const item = analysisProcess.value.find(process => process.step === msg.step) + if (item) { + item.status = msg.status + item.title = msg.title + item.detail = msg.detail + } else { + analysisProcess.value.push({ + step: msg.step, + title: msg.title, + detail: msg.detail, + status: msg.status, + }) + } + } else if (msg.type === 'chunk') { + aiAnalysis.value += msg.content + } else if (msg.type === 'error') { + aiAnalysis.value = msg.message + ElMessage.warning('AI 分析异常') + } else if (msg.type === 'done') { + ElMessage.success('AI 分析完成') + } + } catch (e) { + // ignore + } + } + } } catch (e) { + aiAnalysis.value = 'AI 分析失败: ' + e.message ElMessage.error('AI 分析失败: ' + e.message) } finally { analyzing.value = false @@ -382,18 +470,51 @@ async function sendChat() { chatLoading.value = true await nextTick() scrollChat() + + // 添加一个空的 assistant 消息用于流式填充 + const assistantIdx = chatMessages.value.length + chatMessages.value.push({ role: 'assistant', content: '' }) + try { - const resp = await fetch(`${API}/chat`, { + const resp = await fetch(`${API}/chat/stream`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ - messages: chatMessages.value.map(m => ({ role: m.role, content: m.content })), + messages: chatMessages.value.slice(0, -1).map(m => ({ role: m.role, content: m.content })), }), }) - const data = await resp.json() - chatMessages.value.push({ role: 'assistant', content: data.reply || '无回复' }) + const reader = resp.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + + while (true) { + const { done, value } = await reader.read() + if (done) break + + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() || '' + + for (const line of lines) { + if (!line.startsWith('data: ')) continue + const jsonStr = line.slice(6) + if (!jsonStr) continue + try { + const msg = JSON.parse(jsonStr) + if (msg.type === 'chunk') { + chatMessages.value[assistantIdx].content += msg.content + await nextTick() + scrollChat() + } else if (msg.type === 'error') { + chatMessages.value[assistantIdx].content = msg.message + } + } catch (e) { + // ignore + } + } + } } catch (e) { - chatMessages.value.push({ role: 'assistant', content: '请求失败: ' + e.message }) + chatMessages.value[assistantIdx].content = '请求失败: ' + e.message } finally { chatLoading.value = false await nextTick() @@ -466,6 +587,21 @@ body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC' .res-name { width: 140px; font-size: 13px; color: #606266; text-align: right; flex-shrink: 0; } .ai-card { border: 1px solid #d9ecff; } +.ai-header-meta { display: flex; align-items: center; gap: 8px; } +.analysis-process { padding: 4px 0; } +.process-heading { display: flex; justify-content: space-between; margin-bottom: 10px; color: #606266; font-size: 13px; font-weight: 600; } +.analysis-step-list { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 10px; margin-top: 14px; } +.analysis-step { + display: flex; align-items: center; gap: 10px; min-height: 64px; + padding: 11px 12px; border: 1px solid #ebeef5; border-radius: 8px; background: #fafafa; +} +.analysis-step.status-running { border-color: #a0cfff; background: #ecf5ff; } +.analysis-step.status-done { border-color: #b3e19d; background: #f0f9eb; } +.analysis-step-icon { display: flex; align-items: center; flex-shrink: 0; } +.analysis-step-body { display: flex; flex: 1; min-width: 0; flex-direction: column; gap: 4px; } +.analysis-step-body strong { color: #303133; font-size: 14px; } +.analysis-step-body span { color: #909399; font-size: 12px; line-height: 1.4; } +.ai-loading { display: flex; align-items: center; gap: 8px; color: #409eff; padding: 6px 0 12px; } .ai-content { line-height: 1.8; font-size: 14px; } .ai-content h1, .ai-content h2, .ai-content h3 { margin: 16px 0 8px; } .ai-content code { background: #f5f7fa; padding: 2px 6px; border-radius: 4px; font-size: 13px; } @@ -488,4 +624,10 @@ body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC' .chat-bubble code { background: rgba(0,0,0,.06); padding: 1px 4px; border-radius: 3px; font-size: 13px; } .chat-bubble pre code { background: none; } .chat-input { display: flex; gap: 8px; margin-top: 12px; } + +@media (max-width: 900px) { + .app-header { height: auto; padding: 12px 16px; align-items: flex-start; gap: 12px; } + .header-left, .header-right { flex-wrap: wrap; } + .analysis-step-list { grid-template-columns: 1fr; } +} diff --git a/tests/test_analyze_process.py b/tests/test_analyze_process.py new file mode 100644 index 0000000..ce154b0 --- /dev/null +++ b/tests/test_analyze_process.py @@ -0,0 +1,44 @@ +import asyncio +import json +import unittest +from unittest.mock import patch + +import backend.main as main + + +def parse_sse(event): + assert event.startswith("data: ") + return json.loads(event[6:].strip()) + + +class AnalyzeProcessStreamTest(unittest.TestCase): + def test_analysis_stream_emits_visible_process_steps_before_content(self): + async def fake_stream(report, question=""): + yield "分析结论" + + async def collect_events(): + main._last_diagnosis = {"score": 80} + main._last_report = "diagnosis report" + with patch.object(main, "analyze_with_ai_stream", fake_stream): + response = await main.analyze_stream(question="关注严重问题") + return [parse_sse(event) async for event in response.body_iterator] + + events = asyncio.run(collect_events()) + process_events = [event for event in events if event["type"] == "process"] + + self.assertEqual( + [event["status"] for event in process_events], + ["running", "done", "running", "done", "running", "done", "running", "done"], + ) + self.assertEqual( + [event["title"] for event in process_events[::2]], + ["读取诊断报告", "识别关键风险", "生成修复方案", "整理分析结论"], + ) + first_chunk_index = next(i for i, event in enumerate(events) if event["type"] == "chunk") + final_process_index = max(i for i, event in enumerate(events) if event["type"] == "process") + self.assertLess(final_process_index, first_chunk_index) + self.assertEqual(events[-1]["type"], "done") + + +if __name__ == "__main__": + unittest.main()