2026年9月 我的“听歌识曲”项目实践过程（10）

今天我引入了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, keysession_id, valuejson.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听歌识曲, version1.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(deleteFalse, 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_code500, detailstr(e)) finally: os.unlink(tmp.name) app.post(/recognize) async def recognize(file: UploadFile File(...)): 异步识别提交任务到Kafka 立即返回 session_id session_id str(uuid4()) file_path fdata/uploads/{session_id}.wav os.makedirs(data/uploads, exist_okTrue) 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_code500, detailfKafka 提交失败: {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, host127.0.0.1, port8010, wsnone)运行结果为今天我为系统加上了四道防线死循环检测、工具调用熔断、幻觉防护、数据安全让它从“能跑”进化为“生产可用”。import time import json from collections import defaultdict class DeadLoopDetector: 检测 Agent 死循环 def __init__(self, max_steps25, max_repeats3): 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, threshold5, timeout10): self.failure_count defaultdict(int) self.threshold threshold self.timeout timeout self.circuit_open defaultdict(float) def call_with_retry(self, func, args, retries3): 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} 第 {i1} 次失败: {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_steps5, max_repeats3) try: for i in range(6): detector.check(search_song) except Exception as e: print(f成功拦截死循环: {e}) print(\n 测试 2熔断器 ) breaker CircuitBreaker(threshold3, timeout5) def faulty_tool(): raise Exception(数据库连接超时) try: breaker.call_with_retry(faulty_tool, [], retries5) 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})运行结果为测试通过。