AI大模型智能助手系统架构设计实战

┌─────────────────────────────────────────────────────────┐
│                     接入层 (Gateway)                      │
│   Nginx → API Gateway → 认证鉴权 → 限流熔断 → 审计日志    │
├─────────────────────────────────────────────────────────┤
│                     应用层 (Application)                   │
│   对话管理 │ 工单路由 │ 文档分析 │ 数据查询 │ 知识推荐      │
├─────────────────────────────────────────────────────────┤
│                     推理层 (Inference)                     │
│   Prompt编排 │ 模型路由 │ 流式生成 │ 答案校验 │ 上下文管理  │
├─────────────────────────────────────────────────────────┤
│                     检索层 (Retrieval)                    │
│   向量检索 │ 关键词检索 │ 混合排序 │ 知识图谱 │ Rerank     │
├─────────────────────────────────────────────────────────┤
│                     数据层 (Data)                          │
│   文档解析 │ 分块策略 │ Embedding │ 索引构建 │ 增量同步    │
└─────────────────────────────────────────────────────────┘

基础设施：Redis集群 │ Milvus集群 │ PostgreSQL │ MinIO │ Kubernetes
      

# RAG Pipeline 核心流程
┌──────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────┐
│  用户问题  │ → │ Query预处理   │ → │  多路召回     │ → │  混合排序  │
└──────────┘    │ - 意图识别     │    │ - 向量召回    │    │ - RRF    │
                │ - Query改写   │    │ - BM25召回    │    │ - Rerank │
                │ - 实体抽取    │    │ - KG召回      │    │ - 置信度 │
                │ - 历史上下文   │    │ - Parent-Child│   └──────────┘
                └──────────────┘    └──────────────┘         │
                                                               ▼
┌──────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────┐
│  答案输出  │ ← │ 答案后处理    │ ← │  LLM生成     │ ← │  上下文   │
│ - 流式推送  │    │ - 幻觉检测    │    │ - Prompt模板 │    │  组装    │
│ - 来源引用  │    │ - 敏感脱敏    │    │ - 温度控制    │    │ - Top-K  │
│ - 置信评分  │    │ - 格式校验    │    │ - Token管理   │    │ - 截断   │
└──────────┘    └──────────────┘    └──────────────┘    └──────────┘
      

class HybridRetriever:
    """混合检索器 - 多路召回 + 智能排序"""

    def retrieve(self, query: str, top_k: int = 20) -> list[Document]:
        # 第一级：并行多路召回
        vector_results = self.vector_store.similarity_search(query, k=top_k * 2)
        bm25_results = self.bm25_index.search(query, k=top_k * 2)
        kg_results = self.knowledge_graph.search(query, k=top_k) if has_entities else []

        # 第二级：RRF融合
        fused = reciprocal_rank_fusion(
            [vector_results, bm25_results, kg_results],
            weights=[0.5, 0.3, 0.2]
        )

        # 第三级：Cross-Encoder Rerank
        reranked = self.reranker.rank(query, fused[:30], top_k=top_k)

        return reranked

    def reciprocal_rank_fusion(self, result_lists, weights, k=60):
        """RRF算法 - 将多个排序列表融合为统一排序"""
        scores = defaultdict(float)
        for results, weight in zip(result_lists, weights):
            for rank, doc in enumerate(results):
                scores[doc.id] += weight / (k + rank + 1)
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)
      

# Milvus Collection 设计
collection_schema = {
    "name": "enterprise_knowledge",
    "fields": [
        {"name": "id", "type": "VARCHAR", "max_length": 64, "is_primary": True},
        {"name": "embedding", "type": "FLOAT_VECTOR", "dim": 768},
        {"name": "sparse_embedding", "type": "SPARSE_FLOAT_VECTOR"},  # 稀疏向量用于混合检索
        {"name": "content", "type": "VARCHAR", "max_length": 8192},
        {"name": "metadata", "type": "JSON"},  # 来源、部门、权限、更新时间等
        {"name": "chunk_index", "type": "INT32"},
        {"name": "doc_id", "type": "VARCHAR", "max_length": 64},
    ],
    "index": {
        "field": "embedding",
        "type": "HNSW",         # 高召回率，适合百万级数据
        "params": {"M": 32, "efConstruction": 256},
        "metric": "COSINE"
    }
}
      

# Prompt模板 - 专业技术问答场景
SYSTEM_PROMPT = """
你是一位专业的{domain}领域知识助手。请基于以下检索到的参考资料回答用户问题。

## 回答规范
1. 仅基于提供的参考资料回答，如果资料不足以完整回答，明确说明并提供已知的部分信息
2. 引用来源时使用 [来源X] 标注，X为来源编号
3. 涉及数据、金额、日期等关键信息时，务必核实后引用
4. 如果用户问题涉及多个方面，分点清晰回答
5. 专业术语首次出现时给出简要解释
6. 回答长度控制在300字以内（除非用户要求详细说明）

## 知识上下文
{context}

## 注意事项
- 当前时间：{current_time}
- 用户部门：{user_department}（用于判断信息可见性）
- 保密等级：{security_level}
"""

# 动态Prompt组装策略
class PromptComposer:
    def compose(self, query, retrieved_docs, conversation_history, user_context):
        # 上下文窗口管理：根据模型Token限制动态调整
        max_context_tokens = 4096
        context_chunks = self._select_optimal_chunks(
            retrieved_docs,
            conversation_history,
            max_context_tokens
        )

        # Few-shot示例注入（根据意图匹配）
        few_shot_examples = self.example_store.get_examples(
            intent=detected_intent,
            limit=2
        )

        return self.template.render(
            domain=user_context.domain,
            context=self._format_context(context_chunks),
            examples=few_shot_examples,
            current_time=datetime.now().isoformat(),
            user_department=user_context.department,
            security_level=user_context.security_level
        )
      

# 模型路由策略配置
MODEL_ROUTING_TABLE = {
    "intent:factqa | complexity:low | confidence:high": {
        "primary": "gpt-4o-mini",      # 简单事实问答用轻量模型
        "fallback": "gpt-4o",
        "timeout": 5000,
        "max_tokens": 512
    },
    "intent:factqa | complexity:high | confidence:low": {
        "primary": "gpt-4o",            # 复杂推理用强模型
        "fallback": "claude-3.5-sonnet",
        "timeout": 15000,
        "max_tokens": 2048
    },
    "intent:dataquery": {
        "primary": "gpt-4o",            # SQL生成需要强推理能力
        "fallback": "claude-3.5-sonnet",
        "timeout": 10000,
        "max_tokens": 1024
    },
    "intent:summarize": {
        "primary": "deepseek-v3",       # 长文本摘要性价比高
        "fallback": "gpt-4o-mini",
        "timeout": 30000,
        "max_tokens": 4096
    },
    "intent:casual": {
        "primary": "gpt-4o-mini",       # 闲聊不需要强模型
        "timeout": 3000,
        "max_tokens": 256
    }
}

class ModelRouter:
    """智能模型路由 - 基于意图、复杂度和成本最优决策"""

    async def dispatch(self, request: RouterRequest) -> ModelResponse:
        route_key = self._resolve_route(request)
        config = MODEL_ROUTING_TABLE[route_key]

        # 熔断检查
        if self.circuit_breaker.is_open(config["primary"]):
            model = config["fallback"]
        else:
            model = config["primary"]

        # 并发限流
        await self.rate_limiter.acquire(model)

        # 带降级的调用
        try:
            response = await asyncio.wait_for(
                self.model_client.generate(model, request.prompt, config),
                timeout=config["timeout"] / 1000
            )
            self.metrics.record(model, "success", response.latency)
            return response
        except (TimeoutError, ModelError) as e:
            self.circuit_breaker.record_failure(model)
            self.metrics.record(model, "failure", 0)
            return await self._fallback_dispatch(config["fallback"], request)
      

class SemanticChunker:
    """语义感知分块器"""

    def chunk_document(self, document: Document) -> list[Chunk]:
        # Step 1: 结构解析
        sections = self.structure_parser.parse(document.content)

        chunks = []
        for section in sections:
            # Step 2: 句子级Embedding
            sentences = self.split_sentences(section.content)
            embeddings = self.embedding_model.encode(sentences)

            # Step 3: 基于相似度变化的切分点检测
            split_points = self._detect_split_points(embeddings, threshold=0.75)

            # Step 4: 生成Chunk并建立Parent-Child关系
            child_chunks = self._create_child_chunks(sentences, split_points)
            parent_chunk = ParentChunk(
                content=section.content,
                metadata=section.metadata
            )
            for child in child_chunks:
                child.parent_id = parent_chunk.id

            chunks.extend(child_chunks)
            chunks.append(parent_chunk)

        return chunks

    def _detect_split_points(self, embeddings, threshold):
        """基于相邻句子Embedding的余弦相似度检测语义边界"""
        similarities = cosine_similarity(embeddings[:-1], embeddings[1:])
        split_points = [0]
        for i, sim in enumerate(similarities):
            if sim < threshold:
                split_points.append(i + 1)
        return split_points
      

# Embedding微调 - 对比学习框架
class EmbeddingFineTuner:
    """基于对比学习的Embedding微调器"""

    def train(self, triplets: list[Triplet], epochs=3):
        # triplet = (anchor_query, positive_doc, negative_doc)
        dataloader = self._prepare_dataloader(triplets)

        for epoch in range(epochs):
            for batch in dataloader:
                anchor_emb = self.model(batch.anchor)      # [batch, 768]
                positive_emb = self.model(batch.positive)   # [batch, 768]
                negative_emb = self.model(batch.negative)   # [batch, 768]

                # InfoNCE Loss（对比学习标准损失函数）
                loss = self.info_nce_loss(anchor_emb, positive_emb, negative_emb)
                loss.backward()
                self.optimizer.step()
                self.optimizer.zero_grad()

        # 验证集评估
        metrics = self.evaluate(self.val_triplets)
        return metrics  # {'ndcg@10': 0.89, 'recall@50': 0.93, ...}

    def hard_negative_mining(self, query, positive, candidates, k=5):
        """困难负样本挖掘 - 选Embedding近但不相关的文档"""
        query_emb = self.model.encode(query)
        candidate_embs = self.model.encode(candidates)
        similarities = cosine_similarity(query_emb, candidate_embs)

        # 选Embedding相似度最高的，但人工标注为不相关的
        hard_negatives = []
        for idx in np.argsort(similarities)[::-1]:
            if candidates[idx] not in positive.related_docs:
                hard_negatives.append(candidates[idx])
            if len(hard_negatives) >= k:
                break
        return hard_negatives
      

class ContextWindowManager:
    """动态上下文窗口管理器"""

    def build_context(self, request: ChatRequest) -> str:
        total_budget = request.model_token_limit - 1024  # 预留输出空间

        # 预算分配策略
        budget = {
            'system_prompt': 500,       # 固定
            'recent_history': 1500,     # 最近3轮原文
            'history_summary': 300,     # 历史摘要
            'retrieved_context': total_budget - 2300,  # 剩余给检索结果
        }

        # 1. System Prompt（固定）
        system = self.prompt_template.render()

        # 2. 对话历史处理
        recent = request.history[-3:]  # 最近3轮保留原文
        summary = self._summarize_history(request.history[:-3]) if len(request.history) > 3 else ""

        # 3. 检索结果按相关性排序并截断
        retrieved = self._truncate_by_relevance(
            request.retrieved_docs,
            budget['retrieved_context']
        )

        # 4. 关键信息首尾放置
        context = self._optimize_placement(retrieved)

        return self._assemble(system, recent, summary, context)
      

# Embedding服务核心接口
@router.post("/v1/embeddings")
@rate_limit(limit=1000, window=60)  # 1000 req/min
@cache(ttl=3600, key_fn=lambda r: hash(r.query))
async def create_embeddings(request: EmbeddingRequest):
    """生成文本Embedding向量"""
    start_time = time.perf_counter()

    texts = request.texts if request.texts else [request.text]

    # 批量推理
    embeddings = await embedding_model.encode_async(texts)

    latency = (time.perf_counter() - start_time) * 1000
    metrics.histogram("embedding.latency", latency, tags={"model": request.model})

    return EmbeddingResponse(
        embeddings=[e.tolist() for e in embeddings],
        model=request.model,
        dimensions=len(embeddings[0]),
        latency_ms=round(latency, 2)
    )
      

# 流式输出接口
@router.get("/v1/chat/stream")
async def chat_stream(query: str, session_id: str):
    """流式对话接口 - SSE"""

    async def event_generator():
        # Phase 1: 检索（非流式，发送进度事件）
        yield sse_event({"type": "status", "message": "正在检索知识库..."})
        docs = await retriever.retrieve(query)

        yield sse_event({
            "type": "sources",
            "data": [{"title": d.metadata["title"], "url": d.metadata["url"]} for d in docs[:3]]
        })

        # Phase 2: LLM生成（流式）
        yield sse_event({"type": "status", "message": "正在生成回答..."})
        full_answer = ""
        async for token in llm.stream_generate(query, docs):
            full_answer += token
            yield sse_event({"type": "token", "data": token})

        # Phase 3: 后处理
        yield sse_event({"type": "status", "message": "校验中..."})
        confidence = await answer_verifier.verify(full_answer, docs)

        yield sse_event({
            "type": "complete",
            "data": {
                "answer": full_answer,
                "confidence": confidence.score,
                "warnings": confidence.warnings
            }
        })

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx不缓冲
            "Connection": "keep-alive"
        }
    )
      

# Redis 多重角色配置
REDIS_ROLES = {
    # 1. 会话状态管理 - 存储对话历史
    "session": {
        "db": 0,
        "ttl": 86400 * 7,        # 对话历史保留7天
        "key_pattern": "session:{user_id}:{conv_id}",
        "max_memory": "2GB"
    },

    # 2. 语义缓存 - 相似Query复用
    "semantic_cache": {
        "db": 1,
        "ttl": 3600,             # 缓存1小时
        "key_pattern": "cache:{hash(embedding)}",
        "similarity_threshold": 0.95,  # 余弦相似度阈值
        "max_memory": "4GB"
    },

    # 3. 分布式限流 - Token Bucket算法
    "rate_limiter": {
        "db": 2,
        "key_pattern": "rate:{user_id}",
        "limits": {
            "default": "60/min",
            "premium": "200/min"
        }
    },

    # 4. 任务队列 - 文档处理
    "task_queue": {
        "broker": "redis://localhost:6379/3",
        "backend": "redis://localhost:6379/4",
        "worker_pool": 8
    },

    # 5. 实时指标 - Prometheus前缀缓存
    "metrics_cache": {
        "db": 5,
        "ttl": 10,               # 10秒刷新
        "max_memory": "512MB"
    }
}
      

# V3.0 Agent RAG 核心循环
class RAGAgent:
    """Agent模式的RAG - 支持多步推理和自我纠错"""

    async def run(self, query: str, max_iterations: int = 3) -> AgentResponse:
        context = RetrievedContext(query=query)

        for i in range(max_iterations):
            # Step 1: 规划 - 判断下一步行动
            plan = await self.planner.plan(query, context)

            if plan.action == "generate":
                # 信息充分，直接生成答案
                answer = await self.generator.generate(query, context)
                break

            elif plan.action == "retrieve":
                # 需要补充检索
                new_docs = await self.retrieve(plan.refined_query)
                context.add_docs(new_docs)

            elif plan.action == "use_tool":
                # 需要调用外部工具（数据库查询、API等）
                tool_result = await self.tool_executor.execute(plan.tool_call)
                context.add_tool_result(tool_result)

            elif plan.action == "reflect":
                # 自我反思 - 检查当前信息是否充分
                reflection = await self.reflector.reflect(query, context)
                if reflection.is_sufficient:
                    answer = await self.generator.generate(query, context)
                    break
                else:
                    context.add_feedback(reflection.feedback)

        # Step 2: 最终校验
        verification = await self.verifier.verify(answer, context)

        return AgentResponse(
            answer=answer,
            sources=context.get_source_docs(),
            confidence=verification.score,
            iterations=i + 1,
            tools_called=context.get_tool_calls()
        )