企业级 AI 应用开发实战：构建高可用 RAG 知识助手全栈平台

from fastapi import FastAPI
from pydantic import BaseModel
from langchain.retrievers import EnsembleRetriever
from langchain.elasticsearch import ElasticsearchStore
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.llms import ChatOpenAI
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain.retrievers import ContextualCompressionRetriever

app = FastAPI()

# 初始化 Embedding（本地部署，保障数据隐私）
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5")

# 向量存储（Elasticsearch 8.x）
vector_store = ElasticsearchStore(
    es_url="http://es-cluster:9200",
    index_name="doc_vectors",
    embedding=embeddings,
    distance_strategy="COSINE"
)

# 关键词检索器（基于 ES 的 match 查询）
keyword_retriever = vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 50})

# 向量检索器（更注重语义）
vector_retriever = vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 50})

# 混合检索（Ensemble，权重关键词0.3，向量0.7）
ensemble = EnsembleRetriever(
    retrievers=[keyword_retriever, vector_retriever],
    weights=[0.3, 0.7]
)

# 重排序器（Cross-Encoder，使用 bge-reranker-large）
reranker = CrossEncoderReranker(
    model_name="BAAI/bge-reranker-large",
    top_n=5
)
final_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=ensemble
)

@app.post("/retrieve")
async def retrieve(query: str, tenant_id: str):
    # 多租户过滤（通过 ES filter）
    docs = final_retriever.get_relevant_documents(
        query,
        filter={"term": {"tenant_id": tenant_id}}
    )
    return [{"content": d.page_content, "score": d.metadata.get("relevance_score")} for d in docs]

// Spring Boot 服务 - LLM 路由核心
@Service
public class LLMRouter {
    @Autowired private List<LLMProvider> providers; // 不同厂商实现
    
    public String generate(String prompt, String tenantId) {
        // 根据租户级别、模型负载、成本策略动态选择
        String preferred = getTenantPreferredModel(tenantId);
        LLMProvider provider = providers.stream()
            .filter(p -> p.supports(preferred))
            .findFirst()
            .orElse(providers.get(0)); // 降级
        
        try {
            return provider.invoke(prompt);
        } catch (Exception e) {
            log.warn("主模型故障，切换备用", e);
            return providers.get(1).invoke(prompt); // 自动故障转移
        }
    }
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: rag-service
spec:
  replicas: 3
  strategy:
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      containers:
      - name: rag
        image: rag-service:latest
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "8Gi"
            cpu: "4"
        env:
        - name: ES_HOST
          value: "elasticsearch-cluster"
        - name: REDIS_URL
          valueFrom:
            secretKeyRef:
              name: redis-secret
              key: url
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000

from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

tracer = trace.get_tracer(__name__)
FastAPIInstrumentor.instrument_app(app)

@app.post("/generate")
async def generate(query: str, tenant: str):
    with tracer.start_as_current_span("rag_pipeline") as span:
        span.set_attribute("tenant", tenant)
        # 检索
        docs = await retrieve(query, tenant)
        # 构建prompt
        context = "\n".join([d["content"] for d in docs])
        prompt = f"基于以下资料回答问题：\n{context}\n问题：{query}"
        # 调用LLM
        answer = router.generate(prompt, tenant)
        span.set_attribute("answer_length", len(answer))
        return {"answer": answer}

@EventListener
public void onQueryCompleted(QueryCompletedEvent event) {
    auditLogRepository.save(
        AuditLog.builder()
            .userId(event.getUserId())
            .tenant(event.getTenant())
            .query(event.getQuery())
            .retrievedDocIds(event.getDocIds())
            .answer(event.getAnswer())
            .latencyMs(event.getLatency())
            .build()
    );
}