企业官网建设流程全解析

今天，我引入了Kafka消息队列，将音频识别改造为异步处理架构，实现了“秒回”接口的削峰填谷。

import sys import os import json from confluent_kafka import Producer, Consumer sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) producer = Producer({'bootstrap.servers': '127.0.0.1:9092'}) def submit_audio_task(file_path, session_id): task = {"file_path": file_path, "session_id": session_id} producer.produce('audio_recognize', key=session_id, value=json.dumps(task)) producer.flush() print(f"【Producer】已发送任务: {task}") return f"任务已提交: {session_id}" consumer = Consumer({ 'bootstrap.servers': '127.0.0.1:9092', 'group.id': 'recognize-workers-final', 'auto.offset.reset': 'earliest' }) consumer.subscribe(['audio_recognize']) def process_audio_worker(): """消费者：从队列取任务 执行指纹提取和匹配""" from src.utils.fingerprint import fingerprint_file from src.db.database import get_db from src.db.matcher import match_song from src.utils.cache import save_session print("Worker 已启动 等待任务...") while True: msg = consumer.poll(1.0) if msg is None: continue if msg.error(): print(f"【Consumer 报错】: {msg.error()}") continue task = json.loads(msg.value().decode('utf-8')) print(f"\n【Kafka收到消息】: {task}") print(f"正在处理任务: {task['session_id']}") try: hashes = fingerprint_file(task['file_path']) db = next(get_db()) result = match_song(db, hashes) save_session(task['session_id'], { "status": "completed", "result": result }) print(f"任务完成: {task['session_id']}") except Exception as e: print(f"【Worker 真实报错】: {repr(e)}") save_session(task['session_id'], { "status": "error", "error": str(e) }) if __name__ == '__main__': process_audio_worker()

接着 我改造server.py为异步接口

from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.middleware.cors import CORSMiddleware import shutil, os, tempfile, uvicorn, hashlib from uuid import uuid4 from src.utils.fingerprint import fingerprint_file from src.db.database import init_db, get_db from src.db.matcher import store_fingerprints, match_song from src.utils.cache import get_cached_result, cache_song_result, redis_lock, increment_song_play, get_session from src.utils.kafka_processor import submit_audio_task app = FastAPI(title="听歌识曲", version="1.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) @app.on_event("startup") def startup(): init_db() @app.get("/") def root(): return {"status": "ok", "message": "听歌识曲 Agent API"} @app.post("/fingerprint") async def add_song(file: UploadFile = File(...), song_name: str = "unknown"): """同步接口：提取指纹并存入 MySQL""" tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") shutil.copyfileobj(file.file, tmp) tmp.close() try: hashes = fingerprint_file(tmp.name) db = next(get_db()) song_id = store_fingerprints(db, song_name, hashes, tmp.name) return {"song_id": song_id, "song_name": song_name, "hashes_count": len(hashes)} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: os.unlink(tmp.name) @app.post("/recognize") async def recognize(file: UploadFile = File(...)): """异步识别：提交任务到Kafka 立即返回 session_id""" session_id = str(uuid4()) file_path = f"data/uploads/{session_id}.wav" os.makedirs("data/uploads", exist_ok=True) with open(file_path, "wb") as f: content = await file.read() f.write(content) try: submit_audio_task(file_path, session_id) except Exception as e: raise HTTPException(status_code=500, detail=f"Kafka 提交失败: {str(e)}") return {"session_id": session_id, "status": "processing"} @app.get("/result/{session_id}") def get_result(session_id: str): """前端轮询：拿 session_id 来查结果""" result = get_session(session_id) return result or {"status": "processing"} if __name__ == '__main__': uvicorn.run(app, host="127.0.0.1", port=8010, ws="none")

运行结果为：

今天，我为系统加上了四道防线：死循环检测、工具调用熔断、幻觉防护、数据安全，让它从“能跑”进化为“生产可用”。

import time import json from collections import defaultdict class DeadLoopDetector: """检测 Agent 死循环""" def __init__(self, max_steps=25, max_repeats=3): self.max_steps = max_steps self.max_repeats = max_repeats self.step_count = 0 self.action_history = [] def check(self, action): self.step_count += 1 if self.step_count >= self.max_steps: raise Exception(f"超过最大步数 {self.max_steps}，疑似死循环") self.action_history.append(action) recent = self.action_history[-self.max_repeats:] if len(recent) == self.max_repeats and len(set(recent)) == 1: raise Exception(f"连续重复动作 {self.max_repeats} 次，疑似死循环") print(f" [死循环检测] 通过，当前步数: {self.step_count}") class CircuitBreaker: """熔断器：工具连续失败自动熔断""" def __init__(self, threshold=5, timeout=10): self.failure_count = defaultdict(int) self.threshold = threshold self.timeout = timeout self.circuit_open = defaultdict(float) def call_with_retry(self, func, args, retries=3): name = func.__name__ if self.circuit_open[name] > 0: if time.time() - self.circuit_open[name] < self.timeout: raise Exception(f"工具 {name} 已熔断，请稍后再试") else: self.circuit_open[name] = 0 self.failure_count[name] = 0 for i in range(retries): try: result = func(*args) self.failure_count[name] = 0 return result except Exception as e: self.failure_count[name] += 1 print(f" [熔断器] 工具 {name} 第 {i+1} 次失败: {e}") if self.failure_count[name] >= self.threshold: self.circuit_open[name] = time.time() raise Exception(f"工具 {name} 达到熔断阈值，已熔断！") time.sleep(0.5) raise Exception(f"重试 {retries} 次后仍失败") def validate_output(output, expected_schema): """验证 LLM 输出是否符合预期的 JSON 格式""" try: data = json.loads(output) for key in expected_schema: if key not in data: return False, f"缺少字段: {key}" return True, "验证通过" except json.JSONDecodeError: return False, "JSON 格式错误" class SafeOperations: """安全操作：所有写操作需要确认 + 审计日志""" audit_log = [] @classmethod def safe_delete(cls, db, model, item_id): """安全删除：软删除 + 审计日志""" item = db.query(model).filter(model.id == item_id).first() if not item: raise Exception("记录不存在") cls.audit_log.append({ "action": "delete", "model": str(model), "id": item_id, "time": time.time() }) if hasattr(item, 'is_deleted'): item.is_deleted = True else: db.delete(item) db.commit() print(f" [安全操作] 已安全删除记录 {item_id}，审计日志已记录") if __name__ == '__main__': print("=== 测试 1：死循环检测 ===") detector = DeadLoopDetector(max_steps=5, max_repeats=3) try: for i in range(6): detector.check("search_song") except Exception as e: print(f"成功拦截死循环: {e}") print("\n=== 测试 2：熔断器 ===") breaker = CircuitBreaker(threshold=3, timeout=5) def faulty_tool(): raise Exception("数据库连接超时") try: breaker.call_with_retry(faulty_tool, [], retries=5) except Exception as e: print(f"成功触发熔断: {e}") print("\n=== 测试 3：输出验证（防幻觉） ===") # 测试正确的输出 is_valid, msg = validate_output('{"song": "晴天", "artist": "周杰伦"}', ["song", "artist"]) print(f"正确输出验证: {is_valid}, {msg}") # 测试缺少字段的输出 is_valid, msg = validate_output('{"song": "晴天"}', ["song", "artist"]) print(f"缺失字段验证: {is_valid}, {msg}")

运行结果为：

测试通过。