生产级 AI 产品架构之道：RAG 检索优化、Agent 状态管理与全链路可观测性实践

┌─────────────────────────────────────────────────────────────┐
│                     接入层（Gateway）                       │
│    WebSocket / SSE 流式网关  |  限流熔断（Sentinel）      │
├─────────────────────────────────────────────────────────────┤
│                   编排层（Orchestrator）                    │
│    LangGraph 状态机  |  路由（意图分类）| 记忆管理（Mem0） │
├─────────────────────────────────────────────────────────────┤
│                  能力层（Capabilities）                     │
│   RAG 管道 | 工具执行器（MCP）| 多模态理解（CLIP）        │
├─────────────────────────────────────────────────────────────┤
│                   数据层（Data & Index）                    │
│   向量库（Milvus）| 图数据库（Neo4j）| 语义缓存（Redis）  │
└─────────────────────────────────────────────────────────────┘
         ▲                                                   │
         └─────────── 可观测性（OpenTelemetry + Langfuse）───┘

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer('BAAI/bge-small-en-v1.5')

def semantic_chunk(sentences, threshold=0.65):
    embeddings = model.encode(sentences)
    breaks = [0]
    for i in range(1, len(sentences)):
        sim = np.dot(embeddings[i-1], embeddings[i]) / (np.linalg.norm(embeddings[i-1]) * np.linalg.norm(embeddings[i]))
        if sim < threshold:
            breaks.append(i)
    breaks.append(len(sentences))
    return [" ".join(sentences[breaks[i]:breaks[i+1]]) for i in range(len(breaks)-1)]

from transformers import AutoModelForSequenceClassification, AutoTokenizer

rerank_tokenizer = AutoTokenizer.from_pretrained('BAAI/bge-reranker-v2-m3')
rerank_model = AutoModelForSequenceClassification.from_pretrained('BAAI/bge-reranker-v2-m3').to('cuda')

def rerank(query, docs):
    pairs = [[query, doc] for doc in docs]
    inputs = rerank_tokenizer(pairs, padding=True, truncation=True, return_tensors="pt", max_length=512)
    scores = rerank_model(**inputs).logits.view(-1, ).float().detach().cpu().numpy()
    return sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    step_count: int
    tool_results: dict
    retry_counter: int

builder = StateGraph(AgentState)
builder.add_node("call_llm", call_llm_node)
builder.add_node("execute_tool", tool_node)
builder.add_edge("call_llm", "execute_tool")
builder.add_conditional_edges("execute_tool", should_continue, {"continue": "call_llm", "stop": END})

from pydantic import BaseModel, ValidationError
from datetime import datetime, timedelta

class CreateTicketArgs(BaseModel):
    title: str
    priority: int = 1
    due_date: datetime

def safe_tool_call(raw_json: dict):
    try:
        return CreateTicketArgs(**raw_json)
    except ValidationError as e:
        # 启发式修复：尝试补全缺失字段
        if 'due_date' not in raw_json:
            raw_json['due_date'] = datetime.now() + timedelta(days=3)
        return CreateTicketArgs(**raw_json)

from opentelemetry import trace
from langfuse import Langfuse

tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("rag_query") as span:
    docs = retriever.get_relevant_documents(query)
    span.set_attribute("retriever.k", len(docs))
    response = llm.generate(prompt)
    span.set_attribute("llm.tokens", response.usage.total_tokens)
    span.set_status(StatusCode.OK)

async def stream_llm(prompt):
    async with llm_client.stream(prompt) as stream:
        async for chunk in stream:
            yield f"data: {chunk.text}\n\n"
            await asyncio.sleep(0)  # 让出事件循环