AI应用定制化实战：从Vibe Coding到生产级部署

用户请求 → FastAPI Gateway → 对话管理器（意图识别）→ 
  检索器（向量检索 + BM25 关键词检索 → RRF 融合）→ 
  重排序（Cross-Encoder）→ 上下文构建 → LLM 生成 → 
  后处理（安全过滤+格式校验）→ 返回 + 异步缓存写入

# chunking.py
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import HuggingFaceEmbeddings

embedder = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={"device": "cuda"},
    encode_kwargs={"normalize_embeddings": True}
)

# 语义分块器：基于余弦距离的断点检测
splitter = SemanticChunker(
    embeddings=embedder,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=0.85,  # 相似度低于此阈值则切分
    min_chunk_size=200,
    max_chunk_size=800
)

# 处理 PDF 示例
from unstructured.partition.pdf import partition_pdf
elements = partition_pdf("report.pdf", strategy="hi_res", extract_images_in_pdf=True)
text = "\n".join([el.text for el in elements if el.category not in ["Image", "Table"]])
chunks = splitter.split_text(text)  # 返回 List[str]

# retriever.py
from pymilvus import Collection, AnnSearchRequest, KeywordSearchRequest
from pymilvus import RRFRanker

def hybrid_search(query: str, top_k=20):
    # 1. 生成向量和稀疏向量（BM25）
    dense_emb = embedder.embed_query(query)
    sparse_emb = bm25_encoder.encode_queries(query)  # 使用 bm25s 库
    
    # 2. Milvus 混合检索
    dense_req = AnnSearchRequest([dense_emb], "dense_vector", param={"metric_type": "IP", "params": {"nprobe": 16}}, limit=top_k)
    sparse_req = KeywordSearchRequest([sparse_emb], "sparse_vector", param={"metric_type": "IP"}, limit=top_k)
    
    res = collection.hybrid_search(reqs=[dense_req, sparse_req], ranker=RRFRanker(), limit=top_k, output_fields=["text", "metadata"])
    return res[0]  # 返回前十

# 3. 重排序（Cross-Encoder）
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')  # 多语言重排
pairs = [[query, hit['text']] for hit in results]
scores = reranker.predict(pairs)
final = sorted(zip(results, scores), key=lambda x: x[1], reverse=True)[:5]

# graph.py
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List[dict]   # 对话历史
    query: str
    context: str
    intent: str
    answer: str

def intent_node(state: AgentState):
    # 使用轻量分类器（或小模型）判断意图
    prompt = f"判断用户意图：{state['query']}。选项：[general, follow-up, clarification, new_topic]"
    # 调用 LLM（可用 GPT-3.5-turbo）获得 intent
    return {"intent": "follow-up"}

def retrieve_node(state: AgentState):
    if state['intent'] == 'general':
        # 走常规检索
        docs = hybrid_search(state['query'])
    elif state['intent'] == 'follow-up':
        # 结合上一轮上下文改写查询
        new_q = rewrite_query(state['query'], state['messages'][-2:])
        docs = hybrid_search(new_q)
    # ...
    return {"context": format_docs(docs)}

def generate_node(state: AgentState):
    sys_prompt = build_system_prompt(state['context'], state['intent'])
    resp = llm.invoke([{"role": "system", "content": sys_prompt}] + state['messages'])
    return {"answer": resp}

builder = StateGraph(AgentState)
builder.add_node("intent", intent_node)
builder.add_node("retrieve", retrieve_node)
builder.add_node("generate", generate_node)
builder.set_entry_point("intent")
builder.add_edge("intent", "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)
app = builder.compile()

# system_prompt.yaml
role: "企业财务咨询助手"
style: "专业、严谨，引用来源时注明文档名称和页码"
constraints:
  - "若检索内容不足，直接告知无法回答，禁止编造数据"
  - "回答必须分点，每点不超过50字"
  - "涉及金额时，保留两位小数并添加人民币符号"

import yaml
with open("system_prompt.yaml") as f:
    prompt_config = yaml.safe_load(f)
sys_msg = f"你是{prompt_config['role']}，风格：{prompt_config['style']}。约束：{', '.join(prompt_config['constraints'])}。\n\n参考材料：\n{context}"

import redis
import numpy as np
r = redis.Redis(host='localhost', decode_responses=True)

def cached_generate(query: str):
    # 精确匹配
    if r.exists(f"exact:{query}"):
        return r.get(f"exact:{query}")
    # 语义匹配（遍历缓存键的 embedding，可用 Faiss 加速）
    # 省略实现细节...
    # 未命中则调用 LLM
    ans = llm.invoke(query)
    r.setex(f"exact:{query}", 3600, ans)
    return ans

# Dockerfile
FROM python:3.11-slim
RUN pip install poetry && poetry config virtualenvs.create false
COPY pyproject.toml poetry.lock .
RUN poetry install --no-dev
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

├── app/
│   ├── api/              # FastAPI 路由
│   ├── core/             # 配置、常量
│   ├── retrieval/        # 向量索引、混合检索
│   ├── graph/            # LangGraph 工作流
│   ├── prompt/           # YAML 模板 + 优化器
│   └── utils/            # 缓存、日志
├── scripts/              # 离线索引构建、评估
├── tests/                # 单元测试（pytest）
├── deploy/               # K8s yaml + Dockerfile
└── pyproject.toml

poetry install
export MILVUS_HOST=localhost REDIS_URL=redis://localhost
python scripts/build_index.py --data_dir ./docs
uvicorn app.main:app --reload