实时AI应用开发完全教程
适用人群：后端开发者、AI工程师、全栈工程师
前置要求：Python基础、SQL基础、HTTP/WebSocket基本概念
预计学习时间：40-60小时
目录
- 第一章：实时AI架构设计基础
- 第二章：Redis向量搜索实战
- 第三章：PostgreSQL pgvector扩展深入
- 第四章：实时Embedding管线
- 第五章：流式RAG系统开发
- 第六章：WebSocket实时推理
- 第七章：SSE服务端推送
- 第八章：实时推荐引擎
- 第九章：实时异常检测
- 第十章：边缘计算推理
- 第十一章：实战项目一——实时智能客服系统
- 第十二章：实战项目二——流式数据分析平台
- 附录A：常见问题与排错指南
- 附录B：推荐学习路径
第一章：实时AI架构设计基础
1.1 什么是实时AI应用
实时AI应用是指在用户请求到达后，系统能在亚秒级到秒级延迟内完成AI推理并返回结果的应用。与传统的批处理AI系统不同，实时AI强调：
- 低延迟：P99延迟通常要求在500ms以内
- 高并发：同时处理数千到数万用户的请求
- 持续更新：模型和特征数据需要实时或近实时更新
- 流式交互：支持逐token输出、增量结果推送
典型应用场景包括：实时对话系统、实时推荐、实时风控、实时翻译、实时内容审核等。
1.2 实时AI的核心挑战
| 挑战 | 说明 | 解决策略 |
|---|---|---|
| 推理延迟 | 模型推理计算量大 | 模型量化、批处理、GPU优化 |
| 向量检索延迟 | 百万级向量中快速查找近邻 | 近似最近邻(ANN)算法、索引优化 |
| 数据新鲜度 | 用户行为变化需要即时反映 | 流式特征更新、增量索引 |
| 系统复杂度 | 多组件协同，故障传播 | 断路器、降级策略、幂等设计 |
| 成本控制 | GPU资源昂贵 | 动态扩缩容、请求路由优化 |
1.3 架构模式对比
请求-响应模式（同步）：
用户 → API网关 → 推理服务 → 向量数据库 → 返回结果
适用于简单查询，延迟要求不严格的场景。
流式推送模式（异步）：
用户 → WebSocket/SSE → 推理服务(流式生成) → 逐token推送
适用于对话、内容生成等需要流式体验的场景。
事件驱动模式：
数据源 → 消息队列 → 特征管线 → 向量索引更新
→ 异常检测 → 告警推送
适用于实时推荐、异常检测等持续运行的场景。
1.4 技术栈选型指南
# 推荐技术栈组合
REALTIME_AI_STACK = {
"向量数据库": {
"低延迟场景": "Redis Stack (RedisSearch + RediSearch)",
"大规模持久化": "PostgreSQL + pgvector",
"纯向量场景": "Milvus / Qdrant",
},
"推理框架": {
"Python生态": "FastAPI + vLLM / TGI",
"高性能": "Triton Inference Server",
"边缘部署": "ONNX Runtime / TensorRT",
},
"实时通信": {
"双向交互": "WebSocket (FastAPI / Socket.IO)",
"服务端推送": "SSE (Server-Sent Events)",
},
"流处理": {
"轻量级": "Redis Streams",
"企业级": "Apache Kafka / Apache Flink",
},
"Embedding模型": {
"中文场景": "text2vec-large-chinese / bge-large-zh",
"多语言": "text-embedding-3-small (OpenAI) / e5-large-v2",
"轻量部署": "bge-small-zh-v1.5 / MiniLM",
},
}
第二章：Redis向量搜索实战
2.1 Redis Stack与向量搜索概述
Redis Stack是Redis的扩展版本，内置了RediSearch模块，支持原生向量搜索。其核心优势：
- 内存级速度：向量检索延迟通常在1-5ms
- 混合查询：向量搜索 + 标量过滤 + 全文检索组合
- 数据结构丰富：支持JSON、Set、Stream等多种结构
- 部署简单：单节点即可运行，无需分布式集群
适用场景：实时推荐召回、实时相似搜索、会话级缓存向量。
2.2 环境搭建与配置
# 使用Docker启动Redis Stack
docker run -d \
--name redis-stack \
-p 6379:6379 \
-p 8001:8001 \
-e REDIS_ARGS="--requirepass your_password" \
redis/redis-stack:latest
# Python客户端安装
# pip install redis numpy
import redis
import numpy as np
# 连接Redis
r = redis.Redis(
host="localhost",
port=6379,
password="your_password",
decode_responses=True
)
# 验证连接
print(r.ping()) # True
# 检查RediSearch模块是否加载
modules = r.execute_command("MODULE LIST")
print(modules)
2.3 向量索引创建与管理
from redis.commands.search.field import (
VectorField, TagField, NumericField, TextField
)
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
# 定义索引schema
schema = [
# 向量字段：1536维，使用HNSW算法，余弦相似度
VectorField(
"embedding",
"HNSW",
{
"TYPE": "FLOAT32",
"DIM": 1536,
"DISTANCE_METRIC": "COSINE",
"INITIAL_CAP": 100000,
"M": 16, # HNSW图中每个节点的连接数
"EF_CONSTRUCTION": 200, # 构建时的搜索宽度
},
),
# 标量字段
TagField("category", separator=","),
NumericField("timestamp"),
TextField("content"),
TextField("doc_id"),
]
# 创建索引
try:
r.ft("idx:documents").create_index(
fields=schema,
definition=IndexDefinition(
prefix=["doc:"],
index_type=IndexType.HASH,
),
)
print("索引创建成功")
except Exception as e:
if "Index already exists" in str(e):
print("索引已存在，跳过创建")
else:
raise
# 查看索引信息
info = r.ft("idx:documents").info()
print(f"索引文档数: {info['num_docs']}")
print(f"索引字段: {info['attributes']}")
2.4 向量数据写入与查询
import json
import time
def store_document(doc_id: str, content: str, embedding: list[float],
category: str = ""):
"""存储文档及其向量"""
key = f"doc:{doc_id}"
r.hset(key, mapping={
"doc_id": doc_id,
"content": content,
"embedding": np.array(embedding, dtype=np.float32).tobytes(),
"category": category,
"timestamp": int(time.time()),
})
def search_similar(query_embedding: list[float], top_k: int = 10,
category_filter: str = None) -> list[dict]:
"""向量相似搜索"""
from redis.commands.search.query import Query
# 构建查询
query_vector = np.array(query_embedding, dtype=np.float32).tobytes()
# KNN查询语句
knn_query = f"*=>[KNN {top_k} @embedding $vec AS score]"
q = Query(knn_query).dialect(2)
# 可选：添加标量过滤
if category_filter:
q = Query(knn_query).dialect(2).filter(
f"@category=={{{category_filter}}}"
)
q.return_fields("doc_id", "content", "category", "score")
q.sort_by("score", asc=True) # 余弦距离，越小越相似
q.paging(0, top_k)
# 执行查询
results = r.ft("idx:documents").search(
q, query_params={"vec": query_vector}
)
return [
{
"doc_id": doc.doc_id,
"content": doc.content,
"category": doc.category,
"score": float(doc.score),
}
for doc in results.docs
]
2.5 混合搜索：向量+标量过滤
def hybrid_search(
query_embedding: list[float],
categories: list[str] = None,
time_range: tuple[int, int] = None,
top_k: int = 10,
text_keyword: str = None,
) -> list[dict]:
"""混合搜索：向量 + 分类过滤 + 时间范围 + 关键词"""
from redis.commands.search.query import Query
query_vector = np.array(query_embedding, dtype=np.float32).tobytes()
# 构建过滤条件
filters = []
if categories:
# Tag字段多值过滤
cat_expr = "|".join(categories)
filters.append(f"@category=={{{cat_expr}}}")
if time_range:
start, end = time_range
filters.append(f"@timestamp:[{start} {end}]")
if text_keyword:
filters.append(f"@content:{text_keyword}")
# 组合查询
filter_str = ""
if filters:
filter_str = " ".join(filters) + " "
knn_query = f"{filter_str}=>[KNN {top_k} @embedding $vec AS score]"
q = Query(knn_query).dialect(2)
q.return_fields("doc_id", "content", "category", "score", "timestamp")
q.sort_by("score", asc=True)
q.paging(0, top_k)
results = r.ft("idx:documents").search(
q, query_params={"vec": query_vector}
)
return [
{
"doc_id": doc.doc_id,
"content": doc.content,
"category": doc.category,
"score": float(doc.score),
"timestamp": int(doc.timestamp),
}
for doc in results.docs
]
2.6 性能优化与监控
# 关键配置参数调优
def optimize_hnsw_index():
"""HNSW索引优化建议"""
tips = {
"M": "每个节点的连接数。增大M提升召回率，但增加内存和构建时间。推荐值: 16-64",
"EF_CONSTRUCTION": "构建时搜索范围。越大索引质量越好，但构建越慢。推荐值: 200-500",
"EF_RUNTIME": "查询时搜索范围。可通过运行时参数调整，推荐值: 10-200",
"INITIAL_CAP": "预估文档数量，减少动态扩容开销",
}
return tips
# 运行时调整搜索精度
def search_with_ef(query_embedding, top_k=10, ef_runtime=50):
"""通过EF_RUNTIME参数控制搜索精度/速度权衡"""
query_vector = np.array(query_embedding, dtype=np.float32).tobytes()
q = Query(f"*=>[KNN {top_k} @embedding $vec AS score]").dialect(2)
q.return_fields("doc_id", "content", "score")
q.sort_by("score", asc=True)
# 设置运行时EF参数
results = r.ft("idx:documents").search(
q, query_params={"vec": query_vector}
)
return results
# 监控索引状态
def monitor_index(index_name: str = "idx:documents"):
"""监控Redis索引状态"""
info = r.ft(index_name).info()
stats = {
"文档数量": info.get("num_docs", 0),
"索引内存(MB)": round(
int(info.get("inverted_sz_mb", 0)) +
int(info.get("vector_index_sz_mb", 0)), 2
),
"向量索引大小(MB)": info.get("vector_index_sz_mb", 0),
"索引构建状态": info.get("index_status", "unknown"),
}
# Redis内存使用
memory_info = r.info("memory")
stats["Redis使用内存(MB)"] = round(
int(memory_info.get("used_memory", 0)) / 1024 / 1024, 2
)
stats["Redis峰值内存(MB)"] = round(
int(memory_info.get("used_memory_peak", 0)) / 1024 / 1024, 2
)
return stats
第三章：PostgreSQL pgvector扩展深入
3.1 pgvector安装与配置
-- 安装pgvector扩展（需要PostgreSQL 15+）
CREATE EXTENSION IF NOT EXISTS vector;
-- 验证安装
SELECT * FROM pg_extension WHERE extname = 'vector';
-- 配置参数（postgresql.conf或运行时设置）
-- 增加共享内存用于向量索引构建
SET shared_buffers = '4GB';
SET work_mem = '256MB';
SET maintenance_work_mem = '1GB';
# Python连接pgvector
# pip install psycopg2-binary pgvector sqlalchemy
from sqlalchemy import create_engine, text
from pgvector.psycopg2 import register_vector
import psycopg2
# 连接数据库
conn = psycopg2.connect(
host="localhost",
port=5432,
dbname="ai_db",
user="postgres",
password="your_password",
)
register_vector(conn)
# 创建表
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS documents (
id SERIAL PRIMARY KEY,
doc_id VARCHAR(64) UNIQUE NOT NULL,
content TEXT NOT NULL,
category VARCHAR(32),
embedding vector(1536),
metadata JSONB DEFAULT '{}',
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
""")
conn.commit()
3.2 向量数据类型与操作
import numpy as np
def insert_document(cur, doc_id: str, content: str,
embedding: list[float], category: str = "",
metadata: dict = None):
"""插入文档及向量"""
cur.execute("""
INSERT INTO documents (doc_id, content, category, embedding, metadata)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (doc_id)
DO UPDATE SET
content = EXCLUDED.content,
embedding = EXCLUDED.embedding,
category = EXCLUDED.category,
metadata = EXCLUDED.metadata,
updated_at = NOW()
""", (
doc_id, content, category,
np.array(embedding, dtype=np.float32),
json.dumps(metadata or {}),
))
def search_by_vector(cur, query_embedding: list[float],
top_k: int = 10, category: str = None) -> list[dict]:
"""余弦相似度搜索"""
query_vec = np.array(query_embedding, dtype=np.float32)
if category:
cur.execute("""
SELECT doc_id, content, category,
1 - (embedding <=> %s::vector) AS similarity,
metadata
FROM documents
WHERE category = %s
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_vec, category, query_vec, top_k))
else:
cur.execute("""
SELECT doc_id, content, category,
1 - (embedding <=> %s::vector) AS similarity,
metadata
FROM documents
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (query_vec, query_vec, top_k))
return [
{
"doc_id": row[0],
"content": row[1],
"category": row[2],
"similarity": float(row[3]),
"metadata": row[4],
}
for row in cur.fetchall()
]
# pgvector支持的三种距离操作符：
# <=> 余弦距离 (1 - cosine_similarity)
# <-> L2欧几里得距离
# <#> 内积距离 (负内积)
3.3 索引策略：IVFFlat vs HNSW
-- IVFFlat索引：适合大规模数据，构建速度快
-- lists数量建议：rows / 1000 对于 <1M行，sqrt(rows) 对于 >1M行
CREATE INDEX idx_documents_embedding_ivf
ON documents
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);
-- HNSW索引：查询速度快，召回率高，但构建较慢
CREATE INDEX idx_documents_embedding_hnsw
ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- 查看索引构建进度
SELECT phase, tuples_done, tuples_total
FROM pg_stat_progress_create_index;
-- 运行时调整HNSW搜索参数（会话级别）
SET hnsw.ef_search = 100; -- 默认40，增大提升召回率
-- 调整IVFFlat探查的列表数
SET ivfflat.probes = 10; -- 默认1，增大提升召回率但降低速度
# 索引选择建议
INDEX_GUIDE = {
"IVFFlat": {
"适用场景": "数据量大(>100万)，允许近似搜索",
"优势": "构建快，内存占用较小",
"劣势": "需要先插入数据再建索引，召回率略低",
"参数建议": "lists = sqrt(总行数)，probes = lists的1%-10%",
},
"HNSW": {
"适用场景": "对查询延迟和召回率要求高",
"优势": "查询快，召回率高，支持增量插入",
"劣势": "构建慢，内存占用较大",
"参数建议": "m = 16-64, ef_construction = 200-500, ef_search = 40-200",
},
}
3.4 与传统SQL的联合查询
def advanced_search(
cur,
query_embedding: list[float],
keywords: str = None,
category: str = None,
date_from: str = None,
date_to: str = None,
min_similarity: float = 0.5,
top_k: int = 10,
) -> list[dict]:
"""复杂的混合查询：向量 + 全文 + 标量过滤"""
conditions = []
params = []
# 向量相似度基础查询
base_select = """
SELECT doc_id, content, category,
1 - (embedding <=> %s::vector) AS similarity,
metadata, created_at
FROM documents
"""
params.append(np.array(query_embedding, dtype=np.float32))
# 全文搜索条件（使用tsvector）
if keywords:
conditions.append(
"to_tsvector('chinese', content) @@ plainto_tsquery('chinese', %s)"
)
params.append(keywords)
# 分类过滤
if category:
conditions.append("category = %s")
params.append(category)
# 时间范围
if date_from:
conditions.append("created_at >= %s")
params.append(date_from)
if date_to:
conditions.append("created_at <= %s")
params.append(date_to)
# 相似度阈值
conditions.append(
f"1 - (embedding <=> %s::vector) >= %s"
)
params.append(np.array(query_embedding, dtype=np.float32))
params.append(min_similarity)
# 组装查询
where_clause = " AND ".join(conditions) if conditions else "TRUE"
query = f"""
{base_select}
WHERE {where_clause}
ORDER BY embedding <=> %s::vector
LIMIT %s
"""
params.append(np.array(query_embedding, dtype=np.float32))
params.append(top_k)
cur.execute(query, params)
return [
{
"doc_id": row[0],
"content": row[1],
"category": row[2],
"similarity": float(row[3]),
"metadata": row[4],
"created_at": row[5].isoformat() if row[5] else None,
}
for row in cur.fetchall()
]
3.5 大规模向量数据管理
-- 分区表策略：按时间分区，每分区独立建索引
CREATE TABLE documents_partitioned (
id SERIAL,
doc_id VARCHAR(64) NOT NULL,
content TEXT NOT NULL,
category VARCHAR(32),
embedding vector(1536),
created_at TIMESTAMP DEFAULT NOW()
) PARTITION BY RANGE (created_at);
-- 创建月度分区
CREATE TABLE documents_2024_01 PARTITION OF documents_partitioned
FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE documents_2024_02 PARTITION OF documents_partitioned
FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
-- 为每个分区创建独立索引
CREATE INDEX ON documents_2024_01
USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 200);
CREATE INDEX ON documents_2024_02
USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 200);
-- 批量导入优化
-- 1. 临时禁用索引
-- 2. 使用COPY命令批量导入
-- 3. 重建索引
-- 监控向量索引大小
SELECT
schemaname,
tablename,
indexname,
pg_size_pretty(pg_relation_size(indexrelid)) AS index_size
FROM pg_stat_user_indexes
WHERE indexname LIKE '%embedding%';
第四章：实时Embedding管线
4.1 Embedding模型选型
# 模型选型对比
EMBEDDING_MODELS = {
"text2vec-large-chinese": {
"维度": 1024,
"语言": "中文优化",
"速度": "中等",
"质量": "高",
"部署": "sentence-transformers",
"适用": "中文语义搜索、问答匹配",
},
"bge-large-zh-v1.5": {
"维度": 1024,
"语言": "中文优化",
"速度": "中等",
"质量": "很高",
"部署": "sentence-transformers / FlagEmbedding",
"适用": "中文RAG、知识库检索",
},
"bge-small-zh-v1.5": {
"维度": 512,
"语言": "中文优化",
"速度": "快",
"质量": "中高",
"部署": "轻量级，适合边缘部署",
"适用": "资源受限场景、实时性要求极高",
},
"text-embedding-3-small": {
"维度": 1536,
"语言": "多语言",
"速度": "API调用",
"质量": "高",
"部署": "OpenAI API",
"适用": "多语言场景、快速原型",
},
}
4.2 批量与流式Embedding生成
import asyncio
from typing import AsyncIterator
from sentence_transformers import SentenceTransformer
import numpy as np
class EmbeddingService:
"""Embedding生成服务，支持批量和流式"""
def __init__(self, model_name: str = "BAAI/bge-large-zh-v1.5",
batch_size: int = 32, max_length: int = 512):
self.model = SentenceTransformer(model_name)
self.batch_size = batch_size
self.max_length = max_length
self._dimension = self.model.get_sentence_embedding_dimension()
@property
def dimension(self) -> int:
return self._dimension
def encode_batch(self, texts: list[str]) -> np.ndarray:
"""批量编码"""
return self.model.encode(
texts,
batch_size=self.batch_size,
normalize_embeddings=True,
show_progress_bar=False,
)
async def encode_stream(
self, text_iterator: AsyncIterator[str]
) -> AsyncIterator[tuple[str, np.ndarray]]:
"""流式编码：逐批处理输入流"""
batch = []
async for text in text_iterator:
batch.append(text)
if len(batch) >= self.batch_size:
embeddings = self.encode_batch(batch)
for t, e in zip(batch, embeddings):
yield t, e
batch = []
# 处理剩余
if batch:
embeddings = self.encode_batch(batch)
for t, e in zip(batch, embeddings):
yield t, e
def encode_with_prefix(self, texts: list[str],
prefix: str = "为这个句子生成表示以用于检索相关文章：") -> np.ndarray:
"""带指令前缀的编码（BGE系列推荐用法）"""
prefixed = [f"{prefix}{t}" for t in texts]
return self.encode_batch(prefixed)
4.3 Embedding缓存策略
import hashlib
import json
from functools import lru_cache
class EmbeddingCache:
"""多级缓存Embedding服务"""
def __init__(self, embedding_service: EmbeddingService,
redis_client=None, cache_ttl: int = 86400):
self.service = embedding_service
self.redis = redis_client
self.cache_ttl = cache_ttl
self._local_cache = {} # 进程内LRU缓存
def _cache_key(self, text: str) -> str:
"""生成缓存键"""
text_hash = hashlib.md5(text.encode()).hexdigest()
return f"emb:{text_hash}"
def get_embedding(self, text: str) -> np.ndarray:
"""获取embedding，三级缓存：本地→Redis→模型"""
cache_key = self._cache_key(text)
# L1: 本地缓存
if cache_key in self._local_cache:
return self._local_cache[cache_key]
# L2: Redis缓存
if self.redis:
cached = self.redis.get(cache_key)
if cached:
embedding = np.frombuffer(
bytes.fromhex(cached), dtype=np.float32
)
self._local_cache[cache_key] = embedding
return embedding
# L3: 模型推理
embedding = self.service.encode_batch([text])[0]
# 写回缓存
self._local_cache[cache_key] = embedding
if self.redis:
self.redis.setex(
cache_key,
self.cache_ttl,
embedding.tobytes().hex(),
)
return embedding
def get_embeddings_batch(self, texts: list[str]) -> np.ndarray:
"""批量获取，分离缓存命中与未命中"""
results = [None] * len(texts)
uncached_indices = []
uncached_texts = []
for i, text in enumerate(texts):
cache_key = self._cache_key(text)
# L1
if cache_key in self._local_cache:
results[i] = self._local_cache[cache_key]
continue
# L2
if self.redis:
cached = self.redis.get(cache_key)
if cached:
embedding = np.frombuffer(
bytes.fromhex(cached), dtype=np.float32
)
self._local_cache[cache_key] = embedding
results[i] = embedding
continue
uncached_indices.append(i)
uncached_texts.append(text)
# 批量推理未命中的
if uncached_texts:
new_embeddings = self.service.encode_batch(uncached_texts)
for idx, embedding in zip(uncached_indices, new_embeddings):
results[idx] = embedding
cache_key = self._cache_key(texts[idx])
self._local_cache[cache_key] = embedding
if self.redis:
self.redis.setex(
cache_key, self.cache_ttl,
embedding.tobytes().hex(),
)
return np.array(results)
4.4 多模态Embedding处理
class MultiModalEmbedding:
"""多模态Embedding统一处理"""
def __init__(self):
self.text_model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
# 图像模型需要额外安装
# self.image_model = SentenceTransformer("clip-ViT-B-32")
def embed_text(self, text: str) -> np.ndarray:
"""文本Embedding"""
return self.text_model.encode(text, normalize_embeddings=True)
def embed_image(self, image_path: str) -> np.ndarray:
"""图像Embedding（需要CLIP模型）"""
# from PIL import Image
# image = Image.open(image_path)
# return self.image_model.encode(image)
raise NotImplementedError("需要安装CLIP模型")
def cross_modal_search(self, query_text: str,
image_embeddings: np.ndarray,
top_k: int = 5) -> list[int]:
"""跨模态搜索：用文本搜图片"""
query_vec = self.embed_text(query_text)
# 计算相似度
similarities = np.dot(image_embeddings, query_vec)
# 返回Top-K索引
return np.argsort(similarities)[::-1][:top_k].tolist()
4.5 管线监控与告警
import time
from dataclasses import dataclass, field
from collections import deque
@dataclass
class PipelineMetrics:
"""Embedding管线监控指标"""
total_processed: int = 0
total_errors: int = 0
latency_history: deque = field(default_factory=lambda: deque(maxlen=1000))
cache_hits: int = 0
cache_misses: int = 0
def record_latency(self, latency_ms: float):
self.latency_history.append(latency_ms)
self.total_processed += 1
def record_cache_hit(self):
self.cache_hits += 1
def record_cache_miss(self):
self.cache_misses += 1
@property
def avg_latency_ms(self) -> float:
if not self.latency_history:
return 0
return sum(self.latency_history) / len(self.latency_history)
@property
def p99_latency_ms(self) -> float:
if not self.latency_history:
return 0
sorted_latencies = sorted(self.latency_history)
idx = int(len(sorted_latencies) * 0.99)
return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
@property
def cache_hit_rate(self) -> float:
total = self.cache_hits + self.cache_misses
if total == 0:
return 0
return self.cache_hits / total
def check_alerts(self) -> list[str]:
"""检查是否需要告警"""
alerts = []
if self.p99_latency_ms > 500:
alerts.append(f"P99延迟过高: {self.p99_latency_ms:.0f}ms > 500ms")
if self.total_errors / max(self.total_processed, 1) > 0.05:
alerts.append(f"错误率过高: {self.total_errors}/{self.total_processed}")
if self.cache_hit_rate < 0.3 and self.cache_hits + self.cache_misses > 100:
alerts.append(f"缓存命中率过低: {self.cache_hit_rate:.1%}")
return alerts
第五章：流式RAG系统开发
5.1 RAG基础架构回顾
RAG（Retrieval-Augmented Generation）的核心流程：
用户提问 → 查询改写 → 向量检索 → 上下文组装 → LLM生成 → 流式输出
实时RAG的关键区别在于：
- 检索阶段：亚秒级向量检索，可能需要多路召回
- 生成阶段：流式token输出，首token延迟(TTFT)控制在1秒内
- 反馈阶段：支持用户中断、追问、上下文切换
5.2 流式检索设计
import asyncio
from dataclasses import dataclass
@dataclass
class RetrievalResult:
doc_id: str
content: str
score: float
source: str # 来源标识
class StreamingRAG:
"""流式RAG系统"""
def __init__(self, redis_client, pg_conn, llm_client,
embedding_service):
self.redis = redis_client
self.pg = pg_conn
self.llm = llm_client
self.embedder = embedding_service
async def multi_source_retrieve(
self, query: str, top_k: int = 5
) -> list[RetrievalResult]:
"""多路召回：Redis快速检索 + PostgreSQL精确检索"""
query_embedding = self.embedder.encode_batch([query])[0]
# 并行执行两路检索
redis_task = asyncio.create_task(
self._redis_retrieve(query_embedding, top_k)
)
pg_task = asyncio.create_task(
self._pg_retrieve(query_embedding, top_k)
)
redis_results, pg_results = await asyncio.gather(
redis_task, pg_task
)
# 结果融合（RRF: Reciprocal Rank Fusion）
return self._rrf_merge(redis_results, pg_results, k=60)
async def _redis_retrieve(self, embedding, top_k):
"""Redis快速召回"""
# 参考第二章的search_similar实现
from redis.commands.search.query import Query
import numpy as np
query_vector = np.array(embedding, dtype=np.float32).tobytes()
q = Query(f"*=>[KNN {top_k} @embedding $vec AS score]").dialect(2)
q.return_fields("doc_id", "content", "score")
q.sort_by("score", asc=True)
results = self.redis.ft("idx:documents").search(
q, query_params={"vec": query_vector}
)
return [
RetrievalResult(r.doc_id, r.content, float(r.score), "redis")
for r in results.docs
]
async def _pg_retrieve(self, embedding, top_k):
"""PostgreSQL精确召回"""
import numpy as np
with self.pg.cursor() as cur:
cur.execute("""
SELECT doc_id, content, 1 - (embedding <=> %s::vector) AS score
FROM documents
ORDER BY embedding <=> %s::vector
LIMIT %s
""", (np.array(embedding, dtype=np.float32),
np.array(embedding, dtype=np.float32), top_k))
return [
RetrievalResult(row[0], row[1], float(row[2]), "pg")
for row in cur.fetchall()
]
def _rrf_merge(self, *result_lists, k=60) -> list[RetrievalResult]:
"""RRF排序融合"""
scores = {}
doc_map = {}
for results in result_lists:
for rank, r in enumerate(results):
if r.doc_id not in scores:
scores[r.doc_id] = 0
doc_map[r.doc_id] = r
scores[r.doc_id] += 1 / (k + rank + 1)
sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
return [doc_map[doc_id] for doc_id in sorted_ids]
5.3 上下文窗口管理
class ContextManager:
"""上下文窗口管理器"""
def __init__(self, max_context_tokens: int = 3000,
max_history_turns: int = 5):
self.max_context_tokens = max_context_tokens
self.max_history_turns = max_history_turns
def build_context(
self,
query: str,
retrieved_docs: list[RetrievalResult],
chat_history: list[dict],
system_prompt: str = "",
) -> list[dict]:
"""构建发送给LLM的上下文消息"""
messages = []
# 1. 系统提示
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
# 2. 检索文档上下文
doc_context = self._format_retrieved_docs(retrieved_docs)
messages.append({
"role": "system",
"content": f"以下是相关参考资料，请基于这些内容回答用户问题：\n\n{doc_context}"
})
# 3. 对话历史（限制轮数）
recent_history = chat_history[-self.max_history_turns * 2:]
messages.extend(recent_history)
# 4. 当前用户提问
messages.append({"role": "user", "content": query})
# 5. Token预算检查
messages = self._trim_to_fit(messages)
return messages
def _format_retrieved_docs(self, docs: list[RetrievalResult]) -> str:
"""格式化检索结果"""
parts = []
for i, doc in enumerate(docs, 1):
parts.append(
f"[参考资料{i}] (来源: {doc.source}, 相关度: {doc.score:.3f})\n{doc.content}"
)
return "\n\n".join(parts)
def _trim_to_fit(self, messages: list[dict]) -> list[dict]:
"""确保消息不超过token预算（简化实现）"""
# 简化：按字符估算token数（中文约1.5字/token）
total_chars = sum(len(m["content"]) for m in messages)
estimated_tokens = int(total_chars / 1.5)
if estimated_tokens <= self.max_context_tokens:
return messages
# 从历史消息中裁剪
while estimated_tokens > self.max_context_tokens and len(messages) > 3:
# 移除最早的非系统消息
for i in range(len(messages)):
if messages[i]["role"] != "system":
removed = messages.pop(i)
estimated_tokens -= int(len(removed["content"]) / 1.5)
break
return messages
5.4 流式生成与增量输出
import json
from typing import AsyncIterator
class StreamingGenerator:
"""流式生成器"""
def __init__(self, llm_client):
self.llm = llm_client
async def generate_stream(
self, messages: list[dict]
) -> AsyncIterator[str]:
"""流式生成回答"""
async for chunk in self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
stream=True,
temperature=0.7,
max_tokens=2000,
):
if chunk.choices and chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
async def generate_with_citation(
self, messages: list[dict],
docs: list[RetrievalResult]
) -> AsyncIterator[dict]:
"""带引用标注的流式生成"""
# 在系统提示中要求引用来源
citation_prompt = """
请在回答中标注引用来源，格式为 [来源X]。引用格式示例：
根据[来源1]的内容，...，同时[来源2]指出...
"""
messages.insert(1, {"role": "system", "content": citation_prompt})
full_response = ""
async for chunk in self.generate_stream(messages):
full_response += chunk
yield {
"type": "token",
"content": chunk,
"full_text": full_response,
}
# 解析引用
citations = self._extract_citations(full_response, docs)
yield {
"type": "citations",
"citations": citations,
}
def _extract_citations(self, text: str,
docs: list[RetrievalResult]) -> list[dict]:
"""提取引用标注"""
import re
citations = []
pattern = r'\[来源(\d+)\]'
for match in re.finditer(pattern, text):
idx = int(match.group(1)) - 1
if 0 <= idx < len(docs):
citations.append({
"index": idx + 1,
"doc_id": docs[idx].doc_id,
"source": docs[idx].source,
"score": docs[idx].score,
})
return citations
5.5 RAG质量评估与优化
class RAGEvaluator:
"""RAG系统质量评估"""
def evaluate_retrieval(
self,
queries: list[str],
ground_truth: list[list[str]], # 每个query对应的相关doc_id列表
retriever,
top_k: int = 10,
) -> dict:
"""评估检索质量"""
mrr_scores = []
recall_scores = []
for query, relevant_ids in zip(queries, ground_truth):
results = retriever(query, top_k)
retrieved_ids = [r.doc_id for r in results]
# MRR (Mean Reciprocal Rank)
for rank, doc_id in enumerate(retrieved_ids, 1):
if doc_id in relevant_ids:
mrr_scores.append(1.0 / rank)
break
else:
mrr_scores.append(0.0)
# Recall@K
hits = len(set(retrieved_ids) & set(relevant_ids))
recall_scores.append(hits / len(relevant_ids))
return {
"MRR": sum(mrr_scores) / len(mrr_scores),
"Recall@K": sum(recall_scores) / len(recall_scores),
"num_queries": len(queries),
}
def evaluate_generation(
self,
predictions: list[str],
references: list[str],
) -> dict:
"""评估生成质量（简化版）"""
# 实际项目中应使用更专业的评估指标
metrics = {
"avg_response_length": sum(len(p) for p in predictions) / len(predictions),
"num_responses": len(predictions),
}
# 可以集成ROUGE、BLEU或基于LLM的评估
return metrics
第六章：WebSocket实时推理
6.1 WebSocket协议基础
WebSocket是一种在单个TCP连接上进行全双工通信的协议，特别适合实时AI应用：
- 持久连接：避免HTTP反复建连的开销
- 双向通信：客户端和服务端可随时发送消息
- 低延迟：帧头仅2-14字节，远小于HTTP头
- 适用场景：实时对话、流式推理、协同编辑
6.2 推理服务WebSocket接口设计
# pip install fastapi uvicorn websockets
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.websockets import WebSocketState
import json
import asyncio
from typing import Optional
app = FastAPI()
class InferenceHandler:
"""推理处理器"""
def __init__(self, rag_system):
self.rag = rag_system
self.active_sessions: dict[str, list[dict]] = {}
async def handle_message(self, websocket: WebSocket,
session_id: str, message: dict):
"""处理客户端消息"""
msg_type = message.get("type", "")
if msg_type == "query":
await self._handle_query(websocket, session_id, message)
elif msg_type == "context_update":
await self._handle_context_update(session_id, message)
elif msg_type == "stop":
await self._handle_stop(session_id)
elif msg_type == "ping":
await websocket.send_json({"type": "pong"})
else:
await websocket.send_json({
"type": "error",
"content": f"未知消息类型: {msg_type}",
})
async def _handle_query(self, websocket: WebSocket,
session_id: str, message: dict):
"""处理查询请求"""
query = message.get("content", "")
request_id = message.get("request_id", "")
# 初始化会话历史
if session_id not in self.active_sessions:
self.active_sessions[session_id] = []
try:
# 发送开始信号
await websocket.send_json({
"type": "start",
"request_id": request_id,
})
# 1. 检索
await websocket.send_json({
"type": "status",
"content": "正在检索相关信息...",
})
docs = await self.rag.multi_source_retrieve(query)
await websocket.send_json({
"type": "status",
"content": f"找到 {len(docs)} 条相关文档",
"docs_count": len(docs),
})
# 2. 构建上下文
context = self.rag.context_manager.build_context(
query=query,
retrieved_docs=docs,
chat_history=self.active_sessions[session_id],
)
# 3. 流式生成
full_response = ""
async for chunk in self.rag.generator.generate_stream(context):
full_response += chunk
await websocket.send_json({
"type": "token",
"content": chunk,
"request_id": request_id,
})
# 4. 发送完成信号
await websocket.send_json({
"type": "done",
"request_id": request_id,
"full_response": full_response,
})
# 更新会话历史
self.active_sessions[session_id].append(
{"role": "user", "content": query}
)
self.active_sessions[session_id].append(
{"role": "assistant", "content": full_response}
)
except Exception as e:
await websocket.send_json({
"type": "error",
"content": str(e),
"request_id": request_id,
})
async def _handle_context_update(self, session_id: str,
message: dict):
"""处理上下文更新"""
if session_id in self.active_sessions:
# 更新系统提示或上下文配置
pass
async def _handle_stop(self, session_id: str):
"""处理停止生成请求"""
# 通过取消任务来停止生成
pass
6.3 连接管理与负载均衡
from collections import defaultdict
import uuid
class ConnectionManager:
"""WebSocket连接管理器"""
def __init__(self):
# session_id -> websocket
self.active_connections: dict[str, WebSocket] = {}
# user_id -> [session_ids]
self.user_sessions: dict[str, list[str]] = defaultdict(list)
# 连接元数据
self.connection_meta: dict[str, dict] = {}
async def connect(self, websocket: WebSocket,
user_id: str = None) -> str:
"""接受新连接"""
await websocket.accept()
session_id = str(uuid.uuid4())
self.active_connections[session_id] = websocket
self.connection_meta[session_id] = {
"user_id": user_id,
"connected_at": asyncio.get_event_loop().time(),
"last_active": asyncio.get_event_loop().time(),
"message_count": 0,
}
if user_id:
self.user_sessions[user_id].append(session_id)
return session_id
async def disconnect(self, session_id: str):
"""断开连接"""
if session_id in self.active_connections:
meta = self.connection_meta.get(session_id, {})
user_id = meta.get("user_id")
del self.active_connections[session_id]
if session_id in self.connection_meta:
del self.connection_meta[session_id]
if user_id and session_id in self.user_sessions.get(user_id, []):
self.user_sessions[user_id].remove(session_id)
async def send_to_session(self, session_id: str, message: dict):
"""向特定会话发送消息"""
ws = self.active_connections.get(session_id)
if ws and ws.client_state == WebSocketState.CONNECTED:
try:
await ws.send_json(message)
if session_id in self.connection_meta:
self.connection_meta[session_id]["last_active"] = \
asyncio.get_event_loop().time()
self.connection_meta[session_id]["message_count"] += 1
except Exception:
await self.disconnect(session_id)
async def broadcast(self, message: dict):
"""广播消息到所有连接"""
disconnected = []
for session_id, ws in self.active_connections.items():
try:
await ws.send_json(message)
except Exception:
disconnected.append(session_id)
for sid in disconnected:
await self.disconnect(sid)
def get_stats(self) -> dict:
"""获取连接统计"""
return {
"total_connections": len(self.active_connections),
"unique_users": len(self.user_sessions),
"avg_messages": (
sum(m["message_count"] for m in self.connection_meta.values())
/ max(len(self.connection_meta), 1)
),
}
6.4 流式推理结果推送
class StreamingPusher:
"""流式推送管理器"""
def __init__(self, connection_manager: ConnectionManager):
self.conn_mgr = connection_manager
self._cancel_flags: dict[str, asyncio.Event] = {}
async def push_stream(
self, session_id: str, request_id: str,
token_iterator
):
"""推送流式推理结果"""
cancel_event = asyncio.Event()
self._cancel_flags[request_id] = cancel_event
try:
token_count = 0
async for token_data in token_iterator:
# 检查取消标志
if cancel_event.is_set():
await self.conn_mgr.send_to_session(session_id, {
"type": "cancelled",
"request_id": request_id,
})
break
# 推送token
if isinstance(token_data, str):
await self.conn_mgr.send_to_session(session_id, {
"type": "token",
"content": token_data,
"request_id": request_id,
"index": token_count,
})
else:
token_data["request_id"] = request_id
token_data["index"] = token_count
await self.conn_mgr.send_to_session(session_id, token_data)
token_count += 1
# 控制推送速率，避免客户端过载
if token_count % 10 == 0:
await asyncio.sleep(0.01) # 10ms间隔
# 发送完成信号
if not cancel_event.is_set():
await self.conn_mgr.send_to_session(session_id, {
"type": "done",
"request_id": request_id,
"total_tokens": token_count,
})
finally:
self._cancel_flags.pop(request_id, None)
def cancel(self, request_id: str):
"""取消正在推送的流"""
event = self._cancel_flags.get(request_id)
if event:
event.set()
6.5 错误处理与重连机制
# 客户端重连机制（JavaScript示例代码注释）
"""
// WebSocket客户端重连逻辑
class ReconnectingWebSocket {
constructor(url, options = {}) {
this.url = url;
this.maxRetries = options.maxRetries || 10;
this.baseDelay = options.baseDelay || 1000;
this.maxDelay = options.maxDelay || 30000;
this.retryCount = 0;
this.connect();
}
connect() {
this.ws = new WebSocket(this.url);
this.ws.onopen = () => {
console.log('WebSocket连接成功');
this.retryCount = 0;
};
this.ws.onclose = (event) => {
if (!event.wasClean && this.retryCount < this.maxRetries) {
const delay = Math.min(
this.baseDelay * Math.pow(2, this.retryCount),
this.maxDelay
);
console.log(`将在${delay}ms后重连...`);
setTimeout(() => this.connect(), delay);
this.retryCount++;
}
};
this.ws.onerror = (error) => {
console.error('WebSocket错误:', error);
};
}
send(data) {
if (this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(data));
}
}
}
"""
# 服务端错误处理
@app.websocket("/ws/inference/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
connection_mgr = ConnectionManager() # 实际应为单例
handler = InferenceHandler(rag_system) # 实际应注入
session_id = await connection_mgr.connect(websocket, session_id)
try:
while True:
data = await websocket.receive_json()
await handler.handle_message(websocket, session_id, data)
except WebSocketDisconnect:
await connection_mgr.disconnect(session_id)
except json.JSONDecodeError:
await websocket.send_json({
"type": "error",
"content": "无效的JSON格式",
})
except Exception as e:
await websocket.send_json({
"type": "error",
"content": f"服务器内部错误: {str(e)}",
})
await connection_mgr.disconnect(session_id)
第七章：SSE服务端推送
7.1 SSE vs WebSocket选型
| 特性 | SSE | WebSocket |
|---|---|---|
| 通信方向 | 服务端→客户端单向 | 双向 |
| 协议 | HTTP/1.1或HTTP/2 | 独立协议(ws://) |
| 数据格式 | 纯文本 | 文本或二进制 |
| 自动重连 | 浏览器原生支持 | 需手动实现 |
| 防火墙友好 | 是（标准HTTP） | 可能被拦截 |
| 适用场景 | AI流式输出、通知推送 | 实时对话、协同编辑 |
选择建议：
- AI流式输出（一问一答）→ SSE
- 实时双向对话（支持中断、追问）→ WebSocket
- 简单状态推送（模型训练进度）→ SSE
- 复杂交互（多轮对话+上下文管理）→ WebSocket
7.2 SSE服务端实现
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
import time
app = FastAPI()
async def sse_generator(query: str, session_id: str):
"""SSE事件生成器"""
# 发送连接建立事件
yield f"event: connected\ndata: {json.dumps({'session_id': session_id, 'timestamp': time.time()})}\n\n"
try:
# 阶段1：检索
yield f"event: status\ndata: {json.dumps({'phase': 'retrieval', 'message': '正在检索...'})}\n\n"
docs = await rag_system.multi_source_retrieve(query)
yield f"event: status\ndata: {json.dumps({'phase': 'retrieval_done', 'docs_count': len(docs)})}\n\n"
# 阶段2：流式生成
yield f"event: status\ndata: {json.dumps({'phase': 'generation', 'message': '正在生成回答...'})}\n\n"
context = rag_system.context_manager.build_context(
query=query,
retrieved_docs=docs,
chat_history=session_store.get(session_id, []),
)
full_response = ""
token_index = 0
async for chunk in rag_system.generator.generate_stream(context):
full_response += chunk
token_index += 1
yield f"event: token\ndata: {json.dumps({'content': chunk, 'index': token_index})}\n\n"
# 心跳：每20个token发送一次，防止连接超时
if token_index % 20 == 0:
yield f"event: heartbeat\ndata: {json.dumps({'ts': time.time()})}\n\n"
# 完成
yield f"event: done\ndata: {json.dumps({'full_response': full_response, 'total_tokens': token_index})}\n\n"
except Exception as e:
yield f"event: error\ndata: {json.dumps({'error': str(e)})}\n\n"
@app.post("/api/chat/stream")
async def chat_stream(request: Request):
"""SSE流式聊天接口"""
body = await request.json()
query = body.get("query", "")
session_id = body.get("session_id", str(uuid.uuid4()))
return StreamingResponse(
sse_generator(query, session_id),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Nginx禁用缓冲
},
)
7.3 客户端消费与错误恢复
# Python SSE客户端示例
import httpx
import json
class SSEClient:
"""SSE客户端"""
def __init__(self, url: str, max_retries: int = 5):
self.url = url
self.max_retries = max_retries
self._retry_count = 0
async def consume(self, query: str, session_id: str,
on_token=None, on_done=None, on_error=None):
"""消费SSE流"""
async with httpx.AsyncClient() as client:
async with client.stream(
"POST",
self.url,
json={"query": query, "session_id": session_id},
headers={"Accept": "text/event-stream"},
timeout=120.0,
) as response:
event_type = None
async for line in response.aiter_lines():
line = line.strip()
if line.startswith("event:"):
event_type = line[6:].strip()
elif line.startswith("data:"):
data_str = line[5:].strip()
try:
data = json.loads(data_str)
except json.JSONDecodeError:
data = data_str
if event_type == "token" and on_token:
await on_token(data)
elif event_type == "done" and on_done:
await on_done(data)
elif event_type == "error" and on_error:
await on_error(data)
// JavaScript SSE客户端
async function streamChat(query, sessionId) {
const response = await fetch('/api/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ query, session_id: sessionId }),
});
const reader = response.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(); // 保留未完成的行
let eventType = '';
for (const line of lines) {
if (line.startsWith('event:')) {
eventType = line.slice(6).trim();
} else if (line.startsWith('data:')) {
const data = JSON.parse(line.slice(5).trim());
handleEvent(eventType, data);
}
}
}
}
function handleEvent(type, data) {
switch (type) {
case 'token':
appendToOutput(data.content);
break;
case 'done':
finalizeResponse(data);
break;
case 'error':
showError(data.error);
break;
case 'status':
updateStatus(data.message);
break;
}
}
7.4 SSE在AI流式输出中的应用
# 高级SSE功能：多事件流合并
async def enhanced_sse_generator(query: str, session_id: str):
"""增强版SSE生成器，支持进度、引用、token混合推送"""
# 使用asyncio.Queue合并多个事件源
event_queue = asyncio.Queue()
async def retrieval_worker():
"""检索工作器"""
docs = await rag_system.multi_source_retrieve(query)
await event_queue.put(("docs", {"docs": [
{"doc_id": d.doc_id, "score": d.score, "preview": d.content[:100]}
for d in docs
]}))
return docs
async def generation_worker(docs):
"""生成工作器"""
context = rag_system.context_manager.build_context(
query=query, retrieved_docs=docs,
chat_history=session_store.get(session_id, []),
)
async for chunk in rag_system.generator.generate_stream(context):
await event_queue.put(("token", {"content": chunk}))
await event_queue.put(("done", {}))
# 启动检索任务
docs = await retrieval_worker()
# 启动生成任务
gen_task = asyncio.create_task(generation_worker(docs))
# 持续从队列中取出事件并推送
while True:
try:
event_type, data = await asyncio.wait_for(
event_queue.get(), timeout=60.0
)
yield f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
if event_type == "done":
break
except asyncio.TimeoutError:
# 发送心跳
yield f"event: heartbeat\ndata: {json.dumps({'ts': time.time()})}\n\n"
第八章：实时推荐引擎
8.1 推荐系统架构概述
实时推荐引擎的核心组成：
用户行为 → 行为收集服务 → 特征更新管线 → 向量索引更新
↓
用户请求 → 推荐服务 → 向量召回 → 精排 → 业务过滤 → 推荐结果
↑
实时特征(用户画像、上下文)
关键设计原则：
- 双塔模型：用户塔和物品塔独立计算，物品向量离线预计算
- 实时更新：用户行为即时反映到用户向量
- 多路召回：向量召回 + 热门召回 + 协同过滤，融合排序
- AB实验：支持多策略并行对比
8.2 实时特征工程
from datetime import datetime
from collections import defaultdict
import numpy as np
class RealTimeFeatureStore:
"""实时特征存储"""
def __init__(self, redis_client):
self.redis = redis_client
def record_user_action(self, user_id: str, item_id: str,
action: str, timestamp: float = None):
"""记录用户行为并更新特征"""
timestamp = timestamp or time.time()
# 1. 写入行为日志（Redis Stream）
self.redis.xadd(f"user_actions:{user_id}", {
"item_id": item_id,
"action": action,
"timestamp": str(timestamp),
}, maxlen=10000)
# 2. 更新实时特征
self._update_user_features(user_id, item_id, action, timestamp)
self._update_item_features(item_id, action, timestamp)
def _update_user_features(self, user_id: str, item_id: str,
action: str, timestamp: float):
"""更新用户实时特征"""
key = f"features:user:{user_id}"
# 滑动窗口统计（最近1小时、24小时）
pipe = self.redis.pipeline()
# 最近行为列表（用于计算实时兴趣向量）
pipe.lpush(f"user_recent:{user_id}", item_id)
pipe.ltrim(f"user_recent:{user_id}", 0, 99) # 保留最近100个
# 行为计数
pipe.hincrby(key, f"count:{action}", 1)
pipe.hincrby(key, "count:total", 1)
# 最后活跃时间
pipe.hset(key, "last_active", str(timestamp))
# 活跃时段统计
hour = datetime.fromtimestamp(timestamp).hour
pipe.hincrby(key, f"active_hour:{hour}", 1)
pipe.execute()
def _update_item_features(self, item_id: str, action: str,
timestamp: float):
"""更新物品实时特征"""
key = f"features:item:{item_id}"
pipe = self.redis.pipeline()
pipe.hincrby(key, f"count:{action}", 1)
pipe.hincrby(key, "count:total", 1)
pipe.hset(key, "last_interact", str(timestamp))
# 近期热度分（时间衰减）
decay_score = 1.0 / (1.0 + (time.time() - timestamp) / 3600)
pipe.zincrby("item_hot_score", decay_score, item_id)
pipe.execute()
def get_user_realtime_vector(self, user_id: str) -> np.ndarray:
"""计算用户实时兴趣向量（基于最近行为加权平均）"""
recent_items = self.redis.lrange(f"user_recent:{user_id}", 0, 49)
if not recent_items:
return np.zeros(128) # 默认零向量
# 获取这些物品的向量
embeddings = []
for item_id in recent_items:
item_vec = self.redis.get(f"item_vector:{item_id}")
if item_vec:
embeddings.append(
np.frombuffer(bytes.fromhex(item_vec), dtype=np.float32)
)
if not embeddings:
return np.zeros(128)
# 时间加权平均：越近的行为权重越高
weights = np.array([1.0 / (i + 1) for i in range(len(embeddings))])
weights = weights / weights.sum()
user_vector = np.average(embeddings, axis=0, weights=weights)
return user_vector / np.linalg.norm(user_vector) # L2归一化
8.3 向量召回与排序
class RealtimeRecommender:
"""实时推荐引擎"""
def __init__(self, redis_client, feature_store: RealTimeFeatureStore):
self.redis = redis_client
self.feature_store = feature_store
async def recommend(
self, user_id: str, top_k: int = 20,
candidate_pool_size: int = 200,
context: dict = None,
) -> list[dict]:
"""生成实时推荐"""
# 1. 获取用户实时向量
user_vector = self.feature_store.get_user_realtime_vector(user_id)
# 2. 多路召回
candidates = {}
# 路径1：向量召回
vector_candidates = self._vector_recall(
user_vector, candidate_pool_size
)
for item_id, score in vector_candidates:
candidates[item_id] = {"vector_score": score}
# 路径2：热门召回
hot_candidates = self._hot_recall(50)
for item_id, score in hot_candidates:
if item_id not in candidates:
candidates[item_id] = {"vector_score": 0}
candidates[item_id]["hot_score"] = score
# 路径3：协同过滤召回
cf_candidates = self._collaborative_recall(user_id, 50)
for item_id, score in cf_candidates:
if item_id not in candidates:
candidates[item_id] = {"vector_score": 0}
candidates[item_id]["cf_score"] = score
# 3. 精排
ranked = self._rank_candidates(
user_id, candidates, context
)
# 4. 业务过滤（去重、黑名单等）
filtered = self._apply_filters(user_id, ranked)
# 5. 返回Top-K
return filtered[:top_k]
def _vector_recall(self, user_vector: np.ndarray,
top_k: int) -> list[tuple[str, float]]:
"""向量召回"""
# 使用Redis向量搜索
from redis.commands.search.query import Query
query_vector = user_vector.astype(np.float32).tobytes()
q = Query(f"*=>[KNN {top_k} @embedding $vec AS score]").dialect(2)
q.return_fields("item_id", "score")
q.sort_by("score", asc=True)
results = self.redis.ft("idx:items").search(
q, query_params={"vec": query_vector}
)
return [(r.item_id, float(r.score)) for r in results.docs]
def _hot_recall(self, top_k: int) -> list[tuple[str, float]]:
"""热门召回"""
items = self.redis.zrevrangebyscore(
"item_hot_score", "+inf", "-inf",
start=0, num=top_k, withscores=True,
)
return [(item.decode() if isinstance(item, bytes) else item, score)
for item, score in items]
def _collaborative_recall(self, user_id: str,
top_k: int) -> list[tuple[str, float]]:
"""协同过滤召回（简化：基于相似用户的行为）"""
# 获取相似用户（预计算存储在Redis中）
similar_users = self.redis.zrevrange(
f"similar_users:{user_id}", 0, 9, withscores=True
)
item_scores = defaultdict(float)
for sim_user_id, similarity in similar_users:
recent_items = self.redis.lrange(
f"user_recent:{sim_user_id}", 0, 19
)
for item_id in recent_items:
item_id = item_id.decode() if isinstance(item_id, bytes) else item_id
item_scores[item_id] += similarity
sorted_items = sorted(
item_scores.items(), key=lambda x: x[1], reverse=True
)
return sorted_items[:top_k]
def _rank_candidates(self, user_id: str, candidates: dict,
context: dict = None) -> list[dict]:
"""精排打分"""
ranked = []
for item_id, scores in candidates.items():
# 加权融合各路得分
final_score = (
scores.get("vector_score", 0) * 0.5 +
scores.get("hot_score", 0) * 0.2 +
scores.get("cf_score", 0) * 0.3
)
# 上下文加权
if context:
# 时间衰减
if "time_decay" in context:
final_score *= context["time_decay"]
# 多样性惩罚（与已选结果的相似度）
if "diversity_penalty" in context:
final_score *= (1 - context["diversity_penalty"])
ranked.append({
"item_id": item_id,
"score": final_score,
"scores_breakdown": scores,
})
ranked.sort(key=lambda x: x["score"], reverse=True)
return ranked
def _apply_filters(self, user_id: str,
candidates: list[dict]) -> list[dict]:
"""业务过滤"""
# 获取用户已交互物品
seen_items = set(
self.redis.smembers(f"user_seen:{user_id")
)
seen_items = {
x.decode() if isinstance(x, bytes) else x
for x in seen_items
}
filtered = [
c for c in candidates
if c["item_id"] not in seen_items
]
return filtered
8.4 实时兴趣更新
class InterestTracker:
"""用户实时兴趣追踪"""
def __init__(self, redis_client, embedding_service):
self.redis = redis_client
self.embedder = embedding_service
def update_interest_from_action(
self, user_id: str, item_id: str,
item_content: str, action: str
):
"""根据用户行为更新兴趣向量"""
# 获取物品Embedding
item_embedding = self.embedder.encode_batch([item_content])[0]
# 存储物品向量
self.redis.set(
f"item_vector:{item_id}",
item_embedding.astype(np.float32).tobytes().hex(),
ex=86400 * 7, # 7天过期
)
# 根据行为类型加权更新用户兴趣
action_weights = {
"click": 1.0,
"like": 2.0,
"share": 3.0,
"purchase": 4.0,
"skip": -0.5,
"dislike": -1.0,
}
weight = action_weights.get(action, 1.0)
# 指数移动平均更新用户向量
user_vector_key = f"user_vector:{user_id}"
current = self.redis.get(user_vector_key)
alpha = 0.1 # 学习率
if current:
current_vector = np.frombuffer(
bytes.fromhex(current), dtype=np.float32
)
new_vector = (1 - alpha) * current_vector + alpha * weight * item_embedding
else:
new_vector = item_embedding * weight
# L2归一化
norm = np.linalg.norm(new_vector)
if norm > 0:
new_vector = new_vector / norm
self.redis.set(
user_vector_key,
new_vector.astype(np.float32).tobytes().hex(),
ex=86400 * 30, # 30天过期
)
8.5 推荐结果缓存与刷新
class RecommendationCache:
"""推荐结果缓存"""
def __init__(self, redis_client, recommender: RealtimeRecommender):
self.redis = redis_client
self.recommender = recommender
self.cache_ttl = 300 # 5分钟缓存
async def get_recommendations(
self, user_id: str, top_k: int = 20,
force_refresh: bool = False
) -> list[dict]:
"""获取推荐结果（带缓存）"""
cache_key = f"rec_cache:{user_id}:{top_k}"
# 尝试从缓存获取
if not force_refresh:
cached = self.redis.get(cache_key)
if cached:
return json.loads(cached)
# 缓存未命中，重新生成
results = await self.recommender.recommend(user_id, top_k)
# 写入缓存
self.redis.setex(
cache_key, self.cache_ttl,
json.dumps(results, ensure_ascii=False),
)
return results
async def invalidate_user_cache(self, user_id: str):
"""用户行为变化时使缓存失效"""
pattern = f"rec_cache:{user_id}:*"
cursor = 0
while True:
cursor, keys = self.redis.scan(cursor, match=pattern, count=100)
if keys:
self.redis.delete(*keys)
if cursor == 0:
break
第九章：实时异常检测
9.1 异常检测算法概览
| 算法 | 类型 | 适用场景 | 实时性 |
|---|---|---|---|
| Z-Score | 统计 | 单指标阈值检测 | 极高 |
| IQR | 统计 | 离群值检测 | 极高 |
| Isolation Forest | ML | 多维特征异常 | 高 |
| LSTM Autoencoder | DL | 时序模式异常 | 中 |
| DBSCAN | 聚类 | 密度异常 | 高 |
9.2 流式数据异常检测
import numpy as np
from collections import deque
from dataclasses import dataclass
@dataclass
class AnomalyResult:
is_anomaly: bool
score: float
threshold: float
metric_name: str
timestamp: float
details: dict = None
class StreamingAnomalyDetector:
"""流式异常检测器"""
def __init__(self, window_size: int = 1000,
z_threshold: float = 3.0):
self.window_size = window_size
self.z_threshold = z_threshold
self.windows: dict[str, deque] = {}
self.stats: dict[str, dict] = {}
def detect(self, metric_name: str, value: float,
timestamp: float = None) -> AnomalyResult:
"""检测单个数据点是否异常"""
timestamp = timestamp or time.time()
if metric_name not in self.windows:
self.windows[metric_name] = deque(maxlen=self.window_size)
self.stats[metric_name] = {"mean": 0, "std": 0, "count": 0}
window = self.windows[metric_name]
stats = self.stats[metric_name]
# 如果窗口未满，先积累数据
if len(window) < 30:
window.append(value)
self._update_stats(metric_name)
return AnomalyResult(
is_anomaly=False, score=0, threshold=self.z_threshold,
metric_name=metric_name, timestamp=timestamp,
)
# 计算Z-Score
mean = stats["mean"]
std = max(stats["std"], 1e-10) # 防止除零
z_score = abs(value - mean) / std
is_anomaly = z_score > self.z_threshold
# 更新窗口（即使异常也加入，但可选择排除）
window.append(value)
self._update_stats(metric_name)
return AnomalyResult(
is_anomaly=is_anomaly,
score=z_score,
threshold=self.z_threshold,
metric_name=metric_name,
timestamp=timestamp,
details={
"value": value,
"mean": mean,
"std": std,
"window_size": len(window),
},
)
def _update_stats(self, metric_name: str):
"""更新统计量"""
window = self.windows[metric_name]
values = list(window)
self.stats[metric_name] = {
"mean": np.mean(values),
"std": np.std(values),
"count": len(values),
}
class IsolationForestDetector:
"""基于Isolation Forest的多维异常检测"""
def __init__(self, contamination: float = 0.05,
n_estimators: int = 100):
from sklearn.ensemble import IsolationForest
self.model = IsolationForest(
contamination=contamination,
n_estimators=n_estimators,
random_state=42,
)
self.is_fitted = False
self.buffer = []
self.buffer_size = 1000
def add_sample(self, features: np.ndarray):
"""添加样本到缓冲区"""
self.buffer.append(features)
# 缓冲区满时重新训练
if len(self.buffer) >= self.buffer_size:
self._retrain()
def _retrain(self):
"""重新训练模型"""
X = np.array(self.buffer)
self.model.fit(X)
self.is_fitted = True
# 保留一半旧数据 + 新数据
self.buffer = self.buffer[len(self.buffer) // 2:]
def detect(self, features: np.ndarray) -> AnomalyResult:
"""检测多维特征是否异常"""
if not self.is_fitted:
return AnomalyResult(
is_anomaly=False, score=0, threshold=0,
metric_name="multivariate", timestamp=time.time(),
)
score = self.model.decision_function(features.reshape(1, -1))[0]
prediction = self.model.predict(features.reshape(1, -1))[0]
return AnomalyResult(
is_anomaly=prediction == -1,
score=-score, # 越小越异常，取反
threshold=0,
metric_name="multivariate",
timestamp=time.time(),
details={"raw_score": score},
)
9.3 时序异常检测实战
class TimeSeriesAnomalyDetector:
"""时序异常检测器"""
def __init__(self, seasonality: int = 24):
self.seasonality = seasonality
self.history: dict[str, list[float]] = {}
self.seasonal_pattern: dict[str, np.ndarray] = {}
def detect_with_seasonality(
self, metric_name: str, value: float,
timestamp: float
) -> AnomalyResult:
"""考虑季节性的异常检测"""
if metric_name not in self.history:
self.history[metric_name] = []
self.history[metric_name].append(value)
history = self.history[metric_name]
# 至少需要2个完整周期的数据
if len(history) < self.seasonality * 2:
return AnomalyResult(
is_anomaly=False, score=0, threshold=3.0,
metric_name=metric_name, timestamp=timestamp,
)
# 计算季节性模式
recent = history[-self.seasonality * 2:]
pattern = np.array(recent[:self.seasonality])
# 去季节性后的残差
expected = pattern[len(history) % self.seasonality]
residual = value - expected
# 基于残差的Z-Score
residuals = [
history[i] - pattern[i % self.seasonality]
for i in range(self.seasonality, len(history))
]
if len(residuals) < 10:
return AnomalyResult(
is_anomaly=False, score=0, threshold=3.0,
metric_name=metric_name, timestamp=timestamp,
)
mean_r = np.mean(residuals)
std_r = max(np.std(residuals), 1e-10)
z_score = abs(residual - mean_r) / std_r
return AnomalyResult(
is_anomaly=z_score > 3.0,
score=z_score,
threshold=3.0,
metric_name=metric_name,
timestamp=timestamp,
details={
"value": value,
"expected": expected,
"residual": residual,
"seasonal_index": len(history) % self.seasonality,
},
)
9.4 告警系统设计
from enum import Enum
from dataclasses import dataclass, field
import asyncio
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class Alert:
level: AlertLevel
metric_name: str
message: str
value: float
threshold: float
timestamp: float
resolved: bool = False
alert_id: str = ""
class AlertManager:
"""告警管理器"""
def __init__(self, redis_client, notification_channels: list = None):
self.redis = redis_client
self.channels = notification_channels or []
self.alert_cooldown: dict[str, float] = {} # 告警冷却
self.cooldown_seconds = 300 # 5分钟冷却
async def process_anomaly(self, result: AnomalyResult):
"""处理异常检测结果"""
if not result.is_anomaly:
# 检查是否有未恢复的告警需要恢复
await self._check_recovery(result.metric_name)
return
# 检查冷却期
alert_key = f"{result.metric_name}:{result.timestamp // self.cooldown_seconds}"
if alert_key in self.alert_cooldown:
return
# 确定告警级别
level = self._determine_level(result)
# 创建告警
alert = Alert(
level=level,
metric_name=result.metric_name,
message=self._format_message(result),
value=result.details.get("value", 0) if result.details else 0,
threshold=result.threshold,
timestamp=result.timestamp,
alert_id=str(uuid.uuid4()),
)
# 存储告警
self.redis.xadd("alerts:stream", {
"alert_id": alert.alert_id,
"level": alert.level.value,
"metric": alert.metric_name,
"message": alert.message,
"timestamp": str(alert.timestamp),
}, maxlen=10000)
# 设置冷却
self.alert_cooldown[alert_key] = time.time()
# 发送通知
await self._send_notifications(alert)
def _determine_level(self, result: AnomalyResult) -> AlertLevel:
"""根据异常程度确定告警级别"""
if result.score > result.threshold * 2:
return AlertLevel.CRITICAL
elif result.score > result.threshold * 1.5:
return AlertLevel.WARNING
else:
return AlertLevel.INFO
def _format_message(self, result: AnomalyResult) -> str:
"""格式化告警消息"""
details = result.details or {}
return (
f"指标 [{result.metric_name}] 异常: "
f"当前值={details.get('value', 'N/A')}, "
f"均值={details.get('mean', 'N/A'):.2f}, "
f"Z-Score={result.score:.2f}"
)
async def _send_notifications(self, alert: Alert):
"""发送告警通知"""
for channel in self.channels:
try:
await channel.send(alert)
except Exception as e:
print(f"告警发送失败: {e}")
async def _check_recovery(self, metric_name: str):
"""检查告警恢复"""
# 查找未恢复的告警
alerts = self.redis.xrevrange("alerts:stream", count=100)
for alert_id, fields in alerts:
if (fields.get(b"metric", b"").decode() == metric_name and
fields.get(b"level", b"").decode() != "recovered"):
# 发送恢复通知
recovery = Alert(
level=AlertLevel.INFO,
metric_name=metric_name,
message=f"指标 [{metric_name}] 已恢复正常",
value=0, threshold=0,
timestamp=time.time(),
resolved=True,
)
await self._send_notifications(recovery)
break
第十章：边缘计算推理
10.1 边缘推理概述
边缘计算推理将AI模型部署在靠近数据源的设备上，而非云端数据中心：
优势：
- 极低延迟（<10ms）
- 数据不出设备，保护隐私
- 离线可用
- 减少带宽成本
适用场景：
- 移动端实时翻译
- IoT设备异常检测
- 车载自动驾驶推理
- 工业质检实时判断
挑战：
- 计算资源受限（CPU/内存/功耗）
- 模型需要压缩和优化
- 模型更新和同步
10.2 模型量化与压缩
# 模型量化：将FP32权重转为INT8，减少75%内存，加速推理
# 方法1：使用ONNX Runtime量化
import onnxruntime as ort
from onnxruntime.quantization import quantize_dynamic, QuantType
def quantize_model(input_path: str, output_path: str):
"""动态量化ONNX模型"""
quantize_dynamic(
model_input=input_path,
model_output=output_path,
weight_type=QuantType.QUInt8,
)
print(f"量化完成: {input_path} → {output_path}")
# 方法2：PyTorch量化
import torch
def pytorch_quantize(model, calibration_data=None):
"""PyTorch动态量化"""
quantized_model = torch.quantization.quantize_dynamic(
model,
{torch.nn.Linear, torch.nn.Conv2d},
dtype=torch.qint8,
)
return quantized_model
# 方法3：模型剪枝
def prune_model(model, sparsity: float = 0.5):
"""结构化剪枝"""
import torch.nn.utils.prune as prune
for name, module in model.named_modules():
if isinstance(module, torch.nn.Linear):
prune.l1_unstructured(module, name='weight', amount=sparsity)
prune.remove(module, 'weight')
return model
# 模型大小对比
def compare_model_sizes(original_path: str, quantized_path: str):
"""对比模型大小"""
import os
original_size = os.path.getsize(original_path) / 1024 / 1024
quantized_size = os.path.getsize(quantized_path) / 1024 / 1024
print(f"原始模型: {original_size:.1f} MB")
print(f"量化模型: {quantized_size:.1f} MB")
print(f"压缩比: {original_size / quantized_size:.2f}x")
10.3 ONNX Runtime部署
import onnxruntime as ort
import numpy as np
class OnnxInferenceEngine:
"""ONNX Runtime推理引擎"""
def __init__(self, model_path: str, use_gpu: bool = False):
providers = ['CPUExecutionProvider']
if use_gpu and 'CUDAExecutionProvider' in ort.get_available_providers():
providers.insert(0, 'CUDAExecutionProvider')
# 会话选项优化
sess_options = ort.SessionOptions()
sess_options.graph_optimization_level = (
ort.GraphOptimizationLevel.ORT_ENABLE_ALL
)
sess_options.intra_op_num_threads = 4
sess_options.inter_op_num_threads = 2
# 执行模式
sess_options.execution_mode = ort.ExecutionMode.ORT_PARALLEL
self.session = ort.InferenceSession(
model_path, sess_options, providers=providers,
)
# 获取输入输出信息
self.input_names = [inp.name for inp in self.session.get_inputs()]
self.output_names = [out.name for out in self.session.get_outputs()]
def infer(self, inputs: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
"""执行推理"""
# 确保输入数据类型正确
prepared = {}
for name in self.input_names:
if name in inputs:
data = inputs[name]
# 获取模型期望的数据类型
expected_type = self.session.get_inputs()[0].type
if 'float32' in expected_type:
data = data.astype(np.float32)
elif 'int64' in expected_type:
data = data.astype(np.int64)
prepared[name] = data
results = self.session.run(self.output_names, prepared)
return dict(zip(self.output_names, results))
def benchmark(self, input_data: dict[str, np.ndarray],
num_iterations: int = 100) -> dict:
"""性能基准测试"""
import time
# 预热
for _ in range(10):
self.infer(input_data)
# 测试
latencies = []
for _ in range(num_iterations):
start = time.perf_counter()
self.infer(input_data)
latencies.append((time.perf_counter() - start) * 1000)
return {
"avg_ms": np.mean(latencies),
"p50_ms": np.percentile(latencies, 50),
"p95_ms": np.percentile(latencies, 95),
"p99_ms": np.percentile(latencies, 99),
"min_ms": np.min(latencies),
"max_ms": np.max(latencies),
"throughput_rps": 1000 / np.mean(latencies),
}
10.4 边缘-云协同架构
class EdgeCloudOrchestrator:
"""边缘-云协同推理编排器"""
def __init__(self, edge_engine: OnnxInferenceEngine,
cloud_endpoint: str):
self.edge = edge_engine
self.cloud_endpoint = cloud_endpoint
self.fallback_threshold = 0.5 # 置信度阈值
async def infer(self, input_data: dict) -> dict:
"""协同推理：边缘优先，复杂场景上云"""
# 1. 边缘推理
edge_result = self.edge.infer(input_data)
confidence = self._get_confidence(edge_result)
# 2. 高置信度直接返回
if confidence > self.fallback_threshold:
return {
"result": edge_result,
"source": "edge",
"confidence": confidence,
}
# 3. 低置信度上云推理
cloud_result = await self._cloud_infer(input_data)
return {
"result": cloud_result,
"source": "cloud",
"confidence": self._get_confidence(cloud_result),
"edge_confidence": confidence,
}
def _get_confidence(self, result: dict) -> float:
"""从推理结果中提取置信度"""
# 假设输出包含logits，通过softmax计算置信度
if "logits" in result:
logits = result["logits"]
probs = np.exp(logits) / np.exp(logits).sum()
return float(np.max(probs))
return 1.0
async def _cloud_infer(self, input_data: dict) -> dict:
"""调用云端推理服务"""
import httpx
async with httpx.AsyncClient(timeout=5.0) as client:
# 序列化numpy数组
serialized = {
k: v.tolist() for k, v in input_data.items()
}
response = await client.post(
f"{self.cloud_endpoint}/infer",
json=serialized,
)
return response.json()
# 模型同步管理
class ModelSyncManager:
"""边缘模型同步管理"""
def __init__(self, redis_client, model_dir: str = "./models"):
self.redis = redis_client
self.model_dir = model_dir
async def check_for_update(self, device_id: str,
current_version: str) -> dict:
"""检查是否有模型更新"""
latest = self.redis.hgetall("model:latest")
if not latest:
return {"has_update": False}
latest_version = latest.get(b"version", b"").decode()
if latest_version != current_version:
return {
"has_update": True,
"version": latest_version,
"download_url": latest.get(b"url", b"").decode(),
"checksum": latest.get(b"checksum", b"").decode(),
}
return {"has_update": False}
async def report_device_status(self, device_id: str,
status: dict):
"""上报设备状态"""
self.redis.hset(f"device:{device_id}", mapping={
"model_version": status.get("model_version", ""),
"last_inference": str(time.time()),
"avg_latency_ms": str(status.get("avg_latency_ms", 0)),
"memory_usage_mb": str(status.get("memory_usage_mb", 0)),
})
self.redis.expire(f"device:{device_id}", 86400)
第十一章：实战项目一——实时智能客服系统
11.1 需求分析与架构设计
功能需求：
- 用户发起咨询，系统实时回复
- 基于知识库的问答（RAG）
- 支持多轮对话，上下文理解
- 无法回答时转人工
- 对话质量监控
架构设计：
┌─────────────┐ ┌──────────────┐ ┌─────────────────┐
│ 前端应用 │────→│ API Gateway │────→│ WebSocket服务 │
│ (React/Vue)│←────│ (Nginx) │←────│ (FastAPI) │
└─────────────┘ └──────────────┘ └────────┬────────┘
│
┌──────────────────────────────┼──────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌──────────────┐ ┌──────────────┐
│ 对话管理服务 │ │ RAG检索服务 │ │ LLM生成服务 │
│ (Session Mgr) │ │ (Retriever) │ │ (Generator) │
└───────┬───────┘ └──────┬───────┘ └──────────────┘
│ │
▼ ▼
┌───────────────┐ ┌──────────────┐
│ Redis │ │ 向量数据库 │
│ (会话缓存) │ │ (Redis/PgVec) │
└───────────────┘ └──────────────┘
11.2 知识库构建与索引
import hashlib
from pathlib import Path
class KnowledgeBase:
"""客服知识库管理"""
def __init__(self, redis_client, pg_conn, embedding_service):
self.redis = redis_client
self.pg = pg_conn
self.embedder = embedding_service
def add_document(self, doc_id: str, content: str,
category: str, metadata: dict = None):
"""添加文档到知识库"""
# 文档分块
chunks = self._split_document(content, chunk_size=500, overlap=50)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc_id}_chunk_{i}"
# 生成Embedding
embedding = self.embedder.encode_batch([chunk])[0]
# 写入Redis
self.redis.hset(f"kb:{chunk_id}", mapping={
"doc_id": doc_id,
"chunk_id": chunk_id,
"content": chunk,
"category": category,
"embedding": embedding.astype(np.float32).tobytes(),
"metadata": json.dumps(metadata or {}),
})
# 写入PostgreSQL（持久化）
with self.pg.cursor() as cur:
cur.execute("""
INSERT INTO knowledge_base
(doc_id, chunk_id, content, category, embedding, metadata)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (chunk_id) DO UPDATE SET
content = EXCLUDED.content,
embedding = EXCLUDED.embedding,
category = EXCLUDED.category
""", (
doc_id, chunk_id, chunk, category,
embedding.astype(np.float32),
json.dumps(metadata or {}),
))
self.pg.commit()
def _split_document(self, text: str, chunk_size: int = 500,
overlap: int = 50) -> list[str]:
"""智能分块：按段落和句子分割"""
# 先按段落分割
paragraphs = text.split("\n\n")
chunks = []
current_chunk = ""
for para in paragraphs:
para = para.strip()
if not para:
continue
if len(current_chunk) + len(para) <= chunk_size:
current_chunk += "\n\n" + para if current_chunk else para
else:
if current_chunk:
chunks.append(current_chunk)
# 长段落需要进一步分割
if len(para) > chunk_size:
sentences = para.replace("。", "。\n").split("\n")
current_chunk = ""
for sent in sentences:
if len(current_chunk) + len(sent) <= chunk_size:
current_chunk += sent
else:
if current_chunk:
chunks.append(current_chunk)
current_chunk = sent
else:
current_chunk = para
if current_chunk:
chunks.append(current_chunk)
# 添加重叠
if overlap > 0 and len(chunks) > 1:
overlapped = [chunks[0]]
for i in range(1, len(chunks)):
prev_tail = chunks[i-1][-overlap:]
overlapped.append(prev_tail + chunks[i])
chunks = overlapped
return chunks
def search(self, query: str, top_k: int = 5,
category: str = None) -> list[dict]:
"""知识库检索"""
query_embedding = self.embedder.encode_batch([query])[0]
# Redis快速检索
from redis.commands.search.query import Query
query_vector = query_embedding.astype(np.float32).tobytes()
filter_str = f"@category=={{{category}}} " if category else ""
knn_query = f"{filter_str}=>[KNN {top_k} @embedding $vec AS score]"
q = Query(knn_query).dialect(2)
q.return_fields("doc_id", "chunk_id", "content", "category", "score")
q.sort_by("score", asc=True)
results = self.redis.ft("idx:knowledge_base").search(
q, query_params={"vec": query_vector}
)
return [
{
"doc_id": r.doc_id,
"chunk_id": r.chunk_id,
"content": r.content,
"category": r.category,
"score": float(r.score),
}
for r in results.docs
]
11.3 实时对话引擎
class CustomerServiceEngine:
"""实时智能客服引擎"""
def __init__(self, knowledge_base: KnowledgeBase,
llm_client, redis_client):
self.kb = knowledge_base
self.llm = llm_client
self.redis = redis_client
self.context_manager = ContextManager(
max_context_tokens=3000, max_history_turns=5
)
self.system_prompt = """你是一个专业的客服助手。请遵循以下规则：
1. 基于提供的知识库内容回答问题
2. 如果知识库中没有相关信息，请诚实告知并建议转人工
3. 回答要简洁、专业、友好
4. 如果用户情绪激动，请先安抚再解答
5. 引用知识库内容时请注明来源"""
async def handle_message(
self, session_id: str, user_message: str
) -> AsyncIterator[str]:
"""处理用户消息，流式返回"""
# 1. 意图识别（简化版）
intent = await self._classify_intent(user_message)
if intent == "transfer_human":
yield "正在为您转接人工客服，请稍候...\n"
await self._notify_human_agent(session_id)
return
# 2. 知识库检索
docs = self.kb.search(user_message, top_k=3)
# 3. 构建上下文
history = self._get_session_history(session_id)
context = self.context_manager.build_context(
query=user_message,
retrieved_docs=[
RetrievalResult(
d["doc_id"], d["content"], d["score"], "knowledge_base"
)
for d in docs
],
chat_history=history,
system_prompt=self.system_prompt,
)
# 4. 流式生成
full_response = ""
async for chunk in self._generate_stream(context):
full_response += chunk
yield chunk
# 5. 保存对话历史
self._save_to_history(session_id, user_message, full_response)
# 6. 记录对话质量指标
self._record_metrics(session_id, user_message, full_response, docs)
async def _classify_intent(self, message: str) -> str:
"""意图分类（简化版）"""
transfer_keywords = ["转人工", "人工客服", "找人", "投诉"]
if any(kw in message for kw in transfer_keywords):
return "transfer_human"
return "qa"
async def _generate_stream(self, context: list[dict]) -> AsyncIterator[str]:
"""流式生成回答"""
async for chunk in self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=context,
stream=True,
temperature=0.7,
max_tokens=1000,
):
if chunk.choices and chunk.choices[0].delta.content:
yield chunk.choices[0].delta.content
def _get_session_history(self, session_id: str) -> list[dict]:
"""获取会话历史"""
history_json = self.redis.get(f"session:{session_id}:history")
if history_json:
return json.loads(history_json)
return []
def _save_to_history(self, session_id: str, user_msg: str,
assistant_msg: str):
"""保存对话历史"""
history = self._get_session_history(session_id)
history.append({"role": "user", "content": user_msg})
history.append({"role": "assistant", "content": assistant_msg})
# 保留最近10轮对话
history = history[-20:]
self.redis.setex(
f"session:{session_id}:history",
3600, # 1小时过期
json.dumps(history, ensure_ascii=False),
)
async def _notify_human_agent(self, session_id: str):
"""通知人工客服"""
self.redis.xadd("transfer_queue", {
"session_id": session_id,
"timestamp": str(time.time()),
})
def _record_metrics(self, session_id: str, query: str,
response: str, docs: list[dict]):
"""记录对话指标"""
self.redis.xadd("metrics:conversations", {
"session_id": session_id,
"query_length": str(len(query)),
"response_length": str(len(response)),
"docs_count": str(len(docs)),
"top_score": str(docs[0]["score"]) if docs else "0",
"timestamp": str(time.time()),
}, maxlen=100000)
11.4 多轮对话状态管理
class ConversationStateManager:
"""多轮对话状态管理"""
def __init__(self, redis_client):
self.redis = redis_client
def create_session(self, user_id: str, metadata: dict = None) -> str:
"""创建新会话"""
session_id = str(uuid.uuid4())
self.redis.hset(f"conv:{session_id}", mapping={
"user_id": user_id,
"status": "active",
"created_at": str(time.time()),
"turn_count": "0",
"metadata": json.dumps(metadata or {}),
})
self.redis.expire(f"conv:{session_id}", 7200) # 2小时
return session_id
def get_session(self, session_id: str) -> dict:
"""获取会话信息"""
data = self.redis.hgetall(f"conv:{session_id}")
if not data:
return None
return {
k.decode() if isinstance(k, bytes) else k:
v.decode() if isinstance(v, bytes) else v
for k, v in data.items()
}
def update_turn(self, session_id: str, user_msg: str,
bot_msg: str, metadata: dict = None):
"""更新对话轮次"""
pipe = self.redis.pipeline()
# 递增轮次计数
pipe.hincrby(f"conv:{session_id}", "turn_count", 1)
pipe.hset(f"conv:{session_id}", "last_active", str(time.time()))
# 追加到对话日志
log_entry = {
"turn": self.redis.hget(f"conv:{session_id}", "turn_count"),
"user": user_msg,
"bot": bot_msg,
"timestamp": str(time.time()),
}
if metadata:
log_entry["metadata"] = json.dumps(metadata)
pipe.xadd(f"conv_log:{session_id}", log_entry, maxlen=100)
pipe.execute()
def get_conversation_summary(self, session_id: str) -> str:
"""生成对话摘要（用于长对话上下文压缩）"""
logs = self.redis.xrange(f"conv_log:{session_id}", count=20)
summary_parts = []
for _, fields in logs:
user_msg = fields.get(b"user", b"").decode()
bot_msg = fields.get(b"bot", b"").decode()
summary_parts.append(f"用户: {user_msg}\n助手: {bot_msg}")
return "\n".join(summary_parts)
11.5 完整部署方案
# docker-compose.yml
version: '3.8'
services:
# Redis Stack（向量搜索 + 会话缓存）
redis:
image: redis/redis-stack:latest
ports:
- "6379:6379"
- "8001:8001"
volumes:
- redis_data:/data
command: redis-server --requirepass ${REDIS_PASSWORD} --maxmemory 2gb --maxmemory-policy allkeys-lru
# PostgreSQL + pgvector
postgres:
image: pgvector/pgvector:pg16
ports:
- "5432:5432"
environment:
POSTGRES_DB: customer_service
POSTGRES_USER: ${PG_USER}
POSTGRES_PASSWORD: ${PG_PASSWORD}
volumes:
- pg_data:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
# FastAPI应用
app:
build: .
ports:
- "8000:8000"
environment:
REDIS_URL: redis://:${REDIS_PASSWORD}@redis:6379
DATABASE_URL: postgresql://${PG_USER}:${PG_PASSWORD}@postgres:5432/customer_service
depends_on:
- redis
- postgres
deploy:
replicas: 2
resources:
limits:
memory: 4G
cpus: '2'
# Nginx反向代理
nginx:
image: nginx:latest
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
depends_on:
- app
volumes:
redis_data:
pg_data:
# main.py - 完整应用入口
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import StreamingResponse
import os
app = FastAPI(title="实时智能客服系统")
# 初始化组件（实际应使用依赖注入）
redis_client = redis.Redis(
host=os.getenv("REDIS_HOST", "localhost"),
port=int(os.getenv("REDIS_PORT", 6379)),
password=os.getenv("REDIS_PASSWORD"),
decode_responses=True,
)
# 创建向量索引
# ... (参考第二章索引创建代码)
@app.websocket("/ws/chat/{session_id}")
async def chat_websocket(websocket: WebSocket, session_id: str):
await websocket.accept()
engine = CustomerServiceEngine(knowledge_base, llm_client, redis_client)
state_mgr = ConversationStateManager(redis_client)
try:
while True:
data = await websocket.receive_json()
user_message = data.get("message", "")
# 流式推送回复
full_response = ""
async for chunk in engine.handle_message(session_id, user_message):
full_response += chunk
await websocket.send_json({
"type": "token",
"content": chunk,
})
# 更新状态
state_mgr.update_turn(session_id, user_message, full_response)
await websocket.send_json({"type": "done"})
except WebSocketDisconnect:
pass
@app.post("/api/chat/stream")
async def chat_sse(request: Request):
body = await request.json()
session_id = body.get("session_id", str(uuid.uuid4()))
query = body.get("query", "")
engine = CustomerServiceEngine(knowledge_base, llm_client, redis_client)
async def generate():
async for chunk in engine.handle_message(session_id, query):
yield f"event: token\ndata: {json.dumps({'content': chunk})}\n\n"
yield f"event: done\ndata: {{}}\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
第十二章：实战项目二——流式数据分析平台
12.1 平台架构设计
功能需求：
- 实时数据采集与预处理
- 流式异常检测
- 实时聚合分析
- 动态阈值告警
- 可视化仪表盘
技术架构：
数据源(日志/指标/事件)
│
▼
┌──────────────────┐
│ 数据采集层 │
│ (Redis Streams) │
└────────┬─────────┘
│
┌────┴────┐
▼ ▼
┌────────┐ ┌────────────┐
│实时聚合│ │ 异常检测 │
│ 引擎 │ │ 引擎 │
└───┬────┘ └─────┬──────┘
│ │
▼ ▼
┌──────────────────┐
│ Redis缓存层 │
│ (聚合结果/告警) │
└────────┬─────────┘
│
┌────┴────┐
▼ ▼
┌────────┐ ┌────────┐
│API服务 │ │告警服务 │
└───┬────┘ └───┬────┘
│ │
▼ ▼
┌────────┐ ┌────────┐
│前端仪表│ │通知渠道 │
│ 盘 │ │(钉钉等) │
└────────┘ └────────┘
12.2 数据采集与预处理
import json
import time
from typing import Callable
class DataCollector:
"""实时数据采集器"""
def __init__(self, redis_client):
self.redis = redis_client
self.processors: list[Callable] = []
def add_processor(self, processor: Callable):
"""添加数据处理器"""
self.processors.append(processor)
def ingest(self, source: str, data: dict):
"""摄入一条数据"""
# 添加元数据
enriched = {
**data,
"_source": source,
"_ingest_time": str(time.time()),
"_id": str(uuid.uuid4()),
}
# 写入Redis Stream
stream_key = f"stream:{source}"
self.redis.xadd(stream_key, enriched, maxlen=100000)
# 触发处理器
for processor in self.processors:
try:
processor(enriched)
except Exception as e:
print(f"处理器错误: {e}")
def ingest_batch(self, source: str, data_list: list[dict]):
"""批量摄入"""
pipe = self.redis.pipeline()
for data in data_list:
enriched = {
**data,
"_source": source,
"_ingest_time": str(time.time()),
}
pipe.xadd(f"stream:{source}", enriched, maxlen=100000)
pipe.execute()
class StreamConsumer:
"""Stream消费者"""
def __init__(self, redis_client, consumer_group: str,
consumer_name: str):
self.redis = redis_client
self.group = consumer_group
self.consumer = consumer_name
def ensure_group(self, stream_key: str):
"""确保消费者组存在"""
try:
self.redis.xgroup_create(
stream_key, self.group, id="0", mkstream=True
)
except redis.ResponseError as e:
if "BUSYGROUP" not in str(e):
raise
def consume(self, stream_key: str, count: int = 100,
block_ms: int = 5000) -> list[dict]:
"""消费消息"""
self.ensure_group(stream_key)
messages = self.redis.xreadgroup(
self.group, self.consumer,
{stream_key: ">"},
count=count,
block=block_ms,
)
results = []
for stream, entries in messages:
for msg_id, fields in entries:
# 解码fields
decoded = {
k.decode() if isinstance(k, bytes) else k:
v.decode() if isinstance(v, bytes) else v
for k, v in fields.items()
}
decoded["_msg_id"] = msg_id.decode() if isinstance(msg_id, bytes) else msg_id
results.append(decoded)
# 确认消息
self.redis.xack(stream_key, self.group, msg_id)
return results
def consume_stream(self, stream_key: str):
"""持续消费流"""
while True:
messages = self.consume(stream_key, count=10, block_ms=1000)
for msg in messages:
yield msg
12.3 实时分析引擎
class RealtimeAnalyticsEngine:
"""实时分析引擎"""
def __init__(self, redis_client, anomaly_detector: StreamingAnomalyDetector):
self.redis = redis_client
self.detector = anomaly_detector
self.alert_manager = AlertManager(redis_client)
async def process_metric(self, data: dict):
"""处理一条指标数据"""
metric_name = data.get("metric", "")
value = float(data.get("value", 0))
timestamp = float(data.get("_ingest_time", time.time()))
tags = data.get("tags", {})
# 1. 更新实时聚合
self._update_aggregations(metric_name, value, timestamp, tags)
# 2. 异常检测
anomaly_result = self.detector.detect(metric_name, value, timestamp)
# 3. 处理告警
if anomaly_result.is_anomaly:
await self.alert_manager.process_anomaly(anomaly_result)
def _update_aggregations(self, metric_name: str, value: float,
timestamp: float, tags: dict):
"""更新实时聚合"""
pipe = self.redis.pipeline()
# 1分钟窗口聚合
window_key = f"agg:1m:{metric_name}:{int(timestamp) // 60}"
pipe.hincrby(window_key, "count", 1)
pipe.hincrbyfloat(window_key, "sum", value)
# 更新最大最小值
pipe.eval("""
local key = KEYS[1]
local val = tonumber(ARGV[1])
local cur_max = tonumber(redis.call('hget', key, 'max') or '-inf')
local cur_min = tonumber(redis.call('hget', key, 'min') or 'inf')
if val > cur_max then redis.call('hset', key, 'max', val) end
if val < cur_min then redis.call('hset', key, 'min', val) end
""", 1, window_key, value)
pipe.expire(window_key, 3600) # 1小时后过期
# 按标签聚合
for tag_key, tag_value in tags.items():
tag_agg_key = f"agg:tag:{metric_name}:{tag_key}:{tag_value}"
pipe.hincrby(tag_agg_key, "count", 1)
pipe.hincrbyfloat(tag_agg_key, "sum", value)
pipe.expire(tag_agg_key, 3600)
pipe.execute()
def get_metric_stats(self, metric_name: str,
window_minutes: int = 5) -> dict:
"""获取指标统计"""
current_window = int(time.time()) // 60
total_count = 0
total_sum = 0
max_val = float("-inf")
min_val = float("inf")
for i in range(window_minutes):
window_key = f"agg:1m:{metric_name}:{current_window - i}"
stats = self.redis.hgetall(window_key)
if stats:
count = int(stats.get(b"count", 0))
total_count += count
total_sum += float(stats.get(b"sum", 0))
max_val = max(max_val, float(stats.get(b"max", "-inf")))
min_val = min(min_val, float(stats.get(b"min", "inf")))
if total_count == 0:
return {"count": 0, "avg": 0, "max": 0, "min": 0}
return {
"count": total_count,
"avg": total_sum / total_count,
"max": max_val,
"min": min_val,
"sum": total_sum,
}
12.4 可视化与告警
# API接口：为前端仪表盘提供数据
@app.get("/api/metrics/{metric_name}/stats")
async def get_metric_stats(metric_name: str, window: int = 5):
"""获取指标统计"""
stats = analytics_engine.get_metric_stats(metric_name, window)
return {"metric": metric_name, "window_minutes": window, **stats}
@app.get("/api/metrics/{metric_name}/timeseries")
async def get_timeseries(metric_name: str, minutes: int = 60):
"""获取时间序列数据"""
current_window = int(time.time()) // 60
series = []
for i in range(minutes, 0, -1):
window_key = f"agg:1m:{metric_name}:{current_window - i}"
stats = analytics_engine.redis.hgetall(window_key)
if stats:
count = int(stats.get(b"count", 0))
total = float(stats.get(b"sum", 0))
series.append({
"timestamp": (current_window - i) * 60,
"avg": total / count if count > 0 else 0,
"count": count,
})
return {"metric": metric_name, "series": series}
@app.get("/api/alerts/recent")
async def get_recent_alerts(limit: int = 50):
"""获取最近告警"""
alerts = analytics_engine.redis.xrevrange(
"alerts:stream", count=limit
)
return [
{
"id": fields.get(b"alert_id", b"").decode(),
"level": fields.get(b"level", b"").decode(),
"metric": fields.get(b"metric", b"").decode(),
"message": fields.get(b"message", b"").decode(),
"timestamp": float(fields.get(b"timestamp", 0)),
}
for _, fields in alerts
]
# WebSocket实时推送指标更新
@app.websocket("/ws/metrics")
async def metrics_websocket(websocket: WebSocket):
await websocket.accept()
# 客户端订阅的指标
subscribed_metrics = set()
try:
while True:
# 接收订阅请求
data = await websocket.receive_json()
if data.get("action") == "subscribe":
subscribed_metrics.add(data["metric"])
elif data.get("action") == "unsubscribe":
subscribed_metrics.discard(data["metric"])
# 推送最新数据
for metric in subscribed_metrics:
stats = analytics_engine.get_metric_stats(metric, 1)
await websocket.send_json({
"type": "metric_update",
"metric": metric,
"data": stats,
})
await asyncio.sleep(1) # 1秒刷新间隔
except WebSocketDisconnect:
pass
12.5 性能调优与扩展
# 性能调优清单
PERFORMANCE_CHECKLIST = {
"Redis优化": [
"使用pipeline批量操作，减少RTT",
"合理设置maxmemory-policy（推荐allkeys-lru）",
"监控slow log：CONFIG SET slowlog-log-slower-than 10000",
"避免大key（单个key value不超过10KB）",
"使用SCAN代替KEYS命令",
],
"向量搜索优化": [
"HNSW参数调优：M=16-64, ef_construction=200-500",
"运行时调整ef_search控制精度/速度权衡",
"分批插入数据，避免一次性写入过多",
"监控索引内存使用",
],
"流处理优化": [
"合理设置Stream maxlen防止内存溢出",
"使用Consumer Group实现并行消费",
"批量确认消息减少XACK调用",
"监控pending消息数量",
],
"WebSocket优化": [
"设置合理的ping/pong心跳间隔（30秒）",
"限制单用户连接数",
"使用连接池管理后端连接",
"监控活跃连接数和内存使用",
],
}
# 水平扩展方案
class ScalableConsumer:
"""可水平扩展的消费者"""
def __init__(self, redis_client, num_workers: int = 4):
self.redis = redis_client
self.num_workers = num_workers
async def start(self, stream_key: str, processor: Callable):
"""启动多个消费者工作器"""
tasks = []
for i in range(self.num_workers):
consumer = StreamConsumer(
self.redis,
consumer_group="analytics_group",
consumer_name=f"worker_{i}",
)
task = asyncio.create_task(
self._worker_loop(consumer, stream_key, processor)
)
tasks.append(task)
await asyncio.gather(*tasks)
async def _worker_loop(self, consumer: StreamConsumer,
stream_key: str, processor: Callable):
"""单个工作器循环"""
while True:
try:
messages = consumer.consume(stream_key, count=10, block_ms=1000)
for msg in messages:
await processor(msg)
except Exception as e:
print(f"Worker error: {e}")
await asyncio.sleep(1)
附录A：常见问题与排错指南
Q1: Redis向量搜索返回空结果
可能原因：
- 索引前缀与实际key不匹配
- 向量维度不一致
- 数据类型错误（embedding字段应为bytes）
排查步骤：
# 1. 检查索引是否存在
print(r.ft("idx:documents").info())
# 2. 检查key是否存在
print(r.exists("doc:your_doc_id"))
# 3. 检查数据类型
print(r.type("doc:your_doc_id")) # 应为hash
print(r.hget("doc:your_doc_id", "embedding")) # 应为bytes
# 4. 检查向量维度
embedding = r.hget("doc:your_doc_id", "embedding")
print(len(np.frombuffer(embedding, dtype=np.float32))) # 应与索引DIM一致
Q2: pgvector查询慢
可能原因：
- 缺少向量索引
- 索引参数不合理
- work_mem设置过低
解决方案：
-- 1. 检查是否创建了索引
SELECT * FROM pg_indexes WHERE tablename = 'documents';
-- 2. 检查索引构建进度
SELECT phase, tuples_done, tuples_total
FROM pg_stat_progress_create_index;
-- 3. 调整参数
SET work_mem = '256MB';
SET hnsw.ef_search = 100; -- 增大搜索范围
-- 4. 使用EXPLAIN ANALYZE查看执行计划
EXPLAIN ANALYZE
SELECT * FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;
Q3: WebSocket连接频繁断开
可能原因：
- 负载均衡器超时设置过短
- 缺少心跳机制
- Nginx代理缓冲
解决方案：
# nginx.conf
location /ws/ {
proxy_pass http://backend;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_read_timeout 3600s; # 1小时超时
proxy_send_timeout 3600s;
proxy_buffering off; # 禁用缓冲
}
# 服务端心跳
async def heartbeat(websocket: WebSocket, interval: int = 30):
"""定期发送心跳"""
while True:
try:
await websocket.send_json({"type": "ping"})
await asyncio.sleep(interval)
except Exception:
break
Q4: Embedding缓存命中率低
可能原因：
- 文本未归一化（空格、标点、大小写不一致）
- 缓存TTL过短
- 缓存容量不足
解决方案：
def normalize_text(text: str) -> str:
"""文本归一化"""
import re
text = text.strip()
text = re.sub(r'\s+', ' ', text) # 合并空白
text = text.lower() # 小写
text = re.sub(r'[^\w\s\u4e00-\u9fff]', '', text) # 保留中英文和数字
return text
# 缓存键应基于归一化后的文本
def _cache_key(self, text: str) -> str:
normalized = normalize_text(text)
text_hash = hashlib.md5(normalized.encode()).hexdigest()
return f"emb:{text_hash}"
Q5: 流式RAG首token延迟高
可能原因：
- 向量检索慢
- LLM首token延迟（TTFT）
- 网络往返
优化方案：
- 并行化：检索和历史构建并行执行
- 缓存：热门查询预缓存结果
- 模型选择：选择TTFT低的模型
- 预热：保持连接池温暖
# 预热连接池
async def warmup_connections():
"""预热所有连接"""
# 预热Redis
await redis.ping()
# 预热LLM（发送空请求）
await llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "hi"}],
max_tokens=1,
)
# 预热向量搜索
dummy_vec = np.zeros(1536, dtype=np.float32)
# 执行一次dummy搜索...
Q6: 实时推荐系统冷启动
问题：新用户没有行为数据，无法生成个性化推荐
解决方案：
async def recommend_with_cold_start(
self, user_id: str, top_k: int = 20
) -> list[dict]:
"""带冷启动处理的推荐"""
user_vector = self.feature_store.get_user_realtime_vector(user_id)
# 判断是否为冷启动用户
is_cold = np.linalg.norm(user_vector) < 0.01
if is_cold:
# 冷启动策略
context = self.feature_store.get_user_context(user_id)
if context.get("注册来源"):
# 基于来源推荐热门内容
return self._hot_recall(top_k)
else:
# 探索-利用策略：80%热门 + 20%随机
hot = self._hot_recall(int(top_k * 0.8))
random_items = self._random_recall(int(top_k * 0.2))
return hot + random_items
else:
# 正常推荐流程
return await self.recommend(user_id, top_k)
附录B：推荐学习路径
入门阶段（1-2周）
- 基础概念：理解向量、Embedding、ANN搜索原理
- Redis入门：安装Redis Stack，完成第二章基础操作
- pgvector入门：安装扩展，完成第三章基础操作
- 简单RAG：实现一个基于向量搜索的问答Demo
进阶阶段（2-4周）
- 实时通信：实现WebSocket和SSE接口（第六、七章）
- 流式RAG：实现完整的流式RAG系统（第五章）
- Embedding优化：实现缓存和批处理管线（第四章）
- 项目实战：完成实时智能客服系统（第十一章）
高级阶段（4-6周）
- 推荐系统：实现实时推荐引擎（第八章）
- 异常检测：实现流式异常检测系统（第九章）
- 边缘部署：模型量化和ONNX部署（第十章）
- 综合项目：完成流式数据分析平台（第十二章）
推荐学习资源
- Redis官方文档：https://redis.io/docs/stack/search/
- pgvector GitHub：https://github.com/pgvector/pgvector
- FastAPI文档：https://fastapi.tiangolo.com/
- Sentence Transformers：https://www.sbert.net/
- ONNX Runtime：https://onnxruntime.ai/
本教程完。涵盖实时AI应用开发的核心技术栈，从向量数据库到流式推理，从推荐系统到边缘部署，配有完整的代码示例和两个实战项目。建议按学习路径循序渐进，动手实践每个章节的代码。
