企业级智能AI助手架构设计实战

一、项目概述

1.1 业务背景与核心痛点

企业在数字化转型过程中，客服与内部知识助手面临一个根本性矛盾：用户提问的形式是自然语言，但企业后端是结构化的 RESTful 服务。用户不会说"调用 order-service GET /api/orders/{id}"，他们只会说"帮我查一下我的订单"——但这条查询背后对应的是真实的业务 API 调用。

更复杂的是，用户的提问并不总是能通过 API 解决。有些问题需要 LLM 的推理能力，比如"我上个月消费为什么多了300块"——这需要 LLM 对结构化数据做语义理解和分析。

我们为集团官网构建智能 AI 助手时，面临三个核心挑战：

• 路由难题：如何让模型准确判断一条用户输入该走业务 API 还是 LLM？
• 响应一致性问题：API 返回 JSON，LLM 返回自然语言，如何统一展示给用户？
• 延迟体验：API 响应毫秒级，LLM 响应秒级，用户界面如何优雅处理这个差异？

1.2 项目目标

1.3 关键技术选型

二、技术架构设计

2.1 整体架构图

系统采用五层架构设计，从前端到后端依次为：

┌─────────────────────────────────────────────────────────┐
│                   【第一层】前端交互层                      │
│    对话窗口 + Markdown渲染 + SSE流式展示 + 打字机效果       │
└───────────────────────────┬─────────────────────────────────┘
                            │ HTTP / SSE
┌───────────────────────────▼─────────────────────────────────┐
│                    【第二层】API 网关                        │
│    Nginx: SSL终结 / 限流(令牌桶) / IP黑名单 / JWT鉴权       │
└───────────────────────────┬─────────────────────────────────┘
                            │
┌───────────────────────────▼─────────────────────────────────┐
│                   【第三层】意图路由引擎                      │
│    关键词匹配 → 语义向量检索 → 分类模型 → 决策输出           │
│    输出: { path: "api"|"llm"|"hybrid", confidence }         │
└──────────┬───────────────────────┬──────────────────────────┘
           │ path = api            │ path = llm
┌──────────▼───────┐    ┌──────────▼────────────────────────┐
│ 【第四层A】       │    │ 【第四层B】大模型推理路径              │
│  业务API路径      │    │  Prompt构建 + LLM调用                │
│  RESTful API     │    │  知识库检索(RAG) + 响应合成            │
│  微服务集群        │    └─────────────┬──────────────────────┘
└──────────┬───────┘                  │
           │          ┌────────────────┘
           │   ┌──────▼────────────────────────────────────┐
           └──►│          【第五层】响应聚合层                  │
               │   统一格式(卡片/文本) + 上下文注入             │
               │   安全过滤 + 敏感信息脱敏                     │
               └──────────────┬──────────────────────────────┘
                              │ SSE Stream
                       ┌──────▼──────┐
                       │   前端渲染    │
                       └─────────────┘

2.2 请求处理全链路时序

以一个典型查询为例，展示完整处理流程：

用户提问 "查一下我上个月的订单，总金额是多少"
     │
     ▼
① 意图路由引擎
   ├─ 关键词匹配: "订单"命中 → [+查订单, +金额统计]
   ├─ 语义检索: Top-3 候选意图相似度 [0.92, 0.78, 0.31]
   └─ 决策: intent=查订单(置信度0.92) → path=api
     │
     ▼
② 业务API路径执行
   POST /api/orders/query
   ├─ 参数提取: user_id=当前用户, date_range=上个月
   ├─ 调用订单微服务 → 数据库查询
   └─ 返回: { orders: [...], total_amount: ¥8,520 }
     │
     ▼
③ 响应聚合层
   ├─ 格式化: 结构化JSON → 自然语言卡片
   ├─ 上下文注入: 写入Redis session
   └─ 流式输出: SSE → 前端逐Token展示
     │
     ▼
④ 前端渲染
   "您上个月共下单 12 笔，总金额 ¥8,520" [卡片样式]

三、意图路由引擎（核心模块）

3.1 三级路由策略

路由引擎采用三层漏斗模型，从快到准，逐级筛选：

3.2 L1 精确匹配（关键词引擎）

L1 层是性能最优的路径，适用于企业内高频、标准化的业务查询：

INTENT_PATTERNS = {
    "查订单": {
        "keywords": ["订单", "买了", "下单", "购买记录", "待发货", "已发货"],
        "regex": r"(我的|帮我查).{0,4}(订单|购买|买东西)",
        "api_endpoint": "/api/orders/query",
        "params_extractor": extract_order_params,
        "confidence_boost": 0.95
    },
    "查余额": {
        "keywords": ["余额", "账户余额", "还剩多少", "账户"],
        "api_endpoint": "/api/account/balance",
        "confidence_boost": 0.98
    },
    "退换货": {
        "keywords": ["退货", "退款", "换货", "售后", "投诉"],
        "api_endpoint": "/api/aftersales/create",
        "confidence_boost": 0.90
    }
}

def route_l1(user_input: str) -> Optional[RouteResult]:
    normalized = user_input.lower().strip()
    for intent_name, config in INTENT_PATTERNS.items():
        keyword_hit = any(kw in normalized for kw in config["keywords"])
        regex_hit = re.search(config["regex"], user_input) is not None
        if keyword_hit or regex_hit:
            params = config["params_extractor"](user_input)
            return RouteResult(
                intent=intent_name, path=RoutePath.API,
                confidence=config["confidence_boost"],
                api_endpoint=config["api_endpoint"], params=params
            )
    return None  # 未命中，继续L2

3.3 L2 语义检索（向量匹配）

L1 未命中时，启用 L2 向量语义检索。预将所有业务意图构建为向量索引：

class IntentVectorIndex:
    def build_index(self, intents: List[IntentDefinition]):
        texts = [i.query_examples[0] for i in intents]
        embeddings = self.model.encode(texts, normalize_embeddings=True)
        self.index = faiss.IndexFlatIP(embeddings.shape[1])
        self.index.add(embeddings.astype(np.float32))
        self.intent_db = intents

    def search(self, user_input: str, top_k: int = 3) -> List[IntentMatch]:
        query_vec = self.model.encode([user_input], normalize_embeddings=True)
        scores, indices = self.index.search(query_vec.astype(np.float32), top_k)
        results = []
        for score, idx in zip(scores[0], indices[0]):
            if idx < len(self.intent_db):
                results.append(IntentMatch(intent=self.intent_db[idx], similarity=float(score)))
        return results
# sim > 0.85 → 命中 API  |  sim < 0.7 → 触发 L3

3.4 L3 模型分类（边界兜底）

当 L1/L2 置信度不足时，启用 L3 轻量级分类模型做最终决策：

class IntentClassifier:
    def __init__(self, model_path="/models/intent-classifier-v3"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_path, num_labels=3)
        self.id2label = {0: "api", 1: "llm", 2: "hybrid"}

    @torch.no_grad()
    def classify(self, user_input: str) -> RouteResult:
        inputs = self.tokenizer(user_input, return_tensors="pt", truncation=True, max_length=256)
        logits = self.model(**inputs).logits
        probs = torch.softmax(logits, dim=-1)[0]
        pred_id = torch.argmax(probs).item()
        confidence = probs[pred_id].item()
        if confidence < 0.7 or pred_id == 2:
            return RouteResult(intent="unknown", path=RoutePath.LLM, confidence=0.5, fallback=True)
        return RouteResult(intent=self.id2label[pred_id],
                           path=RoutePath.API if pred_id == 0 else RoutePath.LLM,
                           confidence=confidence)

3.5 路由决策输出

@dataclass
class RouteResult:
    intent: str
    path: RoutePath  # API | LLM | HYBRID
    confidence: float
    api_endpoint: Optional[str] = None
    params: Optional[dict] = None
    fallback: bool = False
    latency_ms: float = 0.0

def route(user_input: str) -> RouteResult:
    t0 = time.perf_counter()
    result = route_l1(user_input)
    if not result:
        candidates = route_l2(user_input)
        if candidates and candidates[0].similarity > 0.85:
            result = candidates_to_result(candidates[0])
        else:
            result = route_l3(user_input)
    result.latency_ms = (time.perf_counter() - t0) * 1000
    return result

3.6 多意图检测与置信度融合

真实用户输入往往不单一路由，比如"帮我查下订单，顺便看看有没有优惠"就包含两个意图：

class MultiIntentDetector:
    def detect(self, user_input: str) -> MultiIntentResult:
        l1_matches = route_l1_all(user_input)
        l2_matches = route_l2_all(user_input, top_k=5)
        all_matches = self._merge_and_rank(l1_matches, l2_matches)
        is_multi = len(all_matches) >= 2 and all_matches[0].similarity - all_matches[1].similarity < 0.3
        return MultiIntentResult(intents=all_matches, is_multi=is_multi,
                               routing_plan=self._build_routing_plan(all_matches))

async def execute_multi_intent(result: MultiIntentResult) -> List[Response]:
    tasks = [api_executor.execute(step) if step.path == RoutePath.API
             else llm_client.generate(step.prompt)
             for step in result.routing_plan]
    return await asyncio.gather(*tasks, return_exceptions=True)

3.7 意图分类器的训练与迭代

路由引擎的效果取决于分类器的准确率，需要建立持续迭代机制：

class IntentClassifierTrainer:
    def prepare_training_data(self):
        logs = redis.lrange("router:logs:7d", 0, -1)
        negative = [{"text": json.loads(l)["user_input"],
                     "label": json.loads(l).get("correct_intent", "llm")}
                    for l in logs
                    if json.loads(l).get("user_feedback") == "bad" or json.loads(l)["confidence"] < 0.6]
        positive = [{"text": json.loads(l)["user_input"],
                     "label": json.loads(l)["intent"]}
                    for l in logs
                    if json.loads(l).get("user_feedback") == "good" and json.loads(l)["confidence"] > 0.85]
        return positive + negative

    def train_and_evaluate(self, train_data, test_data):
        model = SentenceTransformersCrossEncoder("/models/base-encoder")
        args = TrainingArguments(output_dir="/models/intent-classifier-v3",
                                 num_train_epochs=5, per_device_train_batch_size=16,
                                 learning_rate=2e-5, warmup_ratio=0.1,
                                 evaluation_strategy="epoch", load_best_model_at_end=True,
                                 metric_for_best_model="f1")
        trainer = Trainer(model=model, args=args, train_dataset=train_data, eval_dataset=test_data)
        trainer.train()
        return trainer.evaluate()

3.8 路由引擎的可配置化设计

业务意图随业务发展不断变化，路由引擎必须支持配置化，无需改代码即可调整策略：

# config/intent_routing.yaml
version: "2.0"
global:
  default_path: llm; confidence_threshold: 0.85; enable_multi_intent: true
l1:
  enabled: true; priority: 1; timeout_ms: 5
l2:
  enabled: true; priority: 2; model: "qwen-embeddings-v2"; top_k: 3
  threshold: 0.85; medium_threshold: 0.70
l3:
  enabled: true; priority: 3; model_path: "/models/intent-classifier-v3"
  confidence_threshold: 0.70; fallback_to_llm: true
circuit_breaker:
  failure_threshold: 5; success_threshold: 3; timeout_seconds: 30

配置变更通过配置中心（Nacos/Apollo）热更新，路由引擎监听配置变更自动 reload 索引和意图定义，零停机更新。

四、业务API处理路径

4.1 API 执行器设计

路由判定为 API 路径后，由 API Executor 统一执行：

class APIExecutor:
    def __init__(self, client: httpx.AsyncClient, cb: CircuitBreaker):
        self.client = client; self.cb = cb

    async def execute(self, route: RouteResult) -> APIResponse:
        try:
            response = await self.cb.call(
                self.client.post(
                    url=f"http://internal-api-gateway{route.api_endpoint}",
                    json=route.params,
                    timeout=httpx.Timeout(3.0, connect=1.0)
            ))
            response.raise_for_status()
            return APIResponse(status="success", data=response.json(),
                               latency_ms=response.elapsed.total_seconds()*1000)
        except httpx.TimeoutException:
            return await self.fallback_to_llm(route)
        except httpx.HTTPStatusError as e:
            return APIResponse(status="api_error", data={"code": e.response.status_code})

4.2 熔断与降级策略

• 熔断器模式：每个 API 维护独立熔断器，连续失败 5 次 / 失败率 > 50% 则熔断
• 超时降级：API 响应超过 3s 触发降级，转 LLM 生成友好回复
• 重试策略：超时和 5xx 错误自动重试 1 次，幂等接口安全重试
• 降级日志：每次降级完整记录，便于 SLA 分析和微服务优化

五、大模型推理路径

5.1 Prompt 工程设计

LLM 路径的核心是 Prompt 设计，采用分层结构：

SYSTEM_PROMPT = """你是一个专业的企业智能助手"小智"。
要求：1. 基于提供的上下文信息回答用户问题 2. 上下文不足时明确告知
3. 不编造不在上下文中的信息 4. 回答简洁专业 5. 适当使用Markdown格式
【安全底线】不输出政治/宗教/色情内容；不泄露商业机密"""

USER_TEMPLATE = """【对话历史】{conversation_history}
【相关知识上下文】{context_chunks}
【用户当前问题】{user_input}"""

5.2 Advanced RAG Pipeline

为防止 LLM 产生幻觉（Hallucination），在调用 LLM 前先从企业知识库检索相关片段：

Advanced RAG Pipeline 全流程:

用户问题: "公司年假政策是怎么规定的？"

① Query 改写（Query Rewriting）
   ├─ HyDE: LLM先生成"假设性答案" → 用答案找相似文档（召回率+15%）
   └─ Query Decomposition: 拆分为子问题 "年假多少天？" + "谁有资格休？"

② 检索增强（Retrieval）
   ├─ 向量检索 (Qwen-Embeddings + FAISS HNSW) top_k=20
   └─ 关键词检索 (BM25) top_k=20

③ RRF 融合排序 (Reciprocal Rank Fusion)
   RRF_score = 1/(k+rank1) + 1/(k+rank2)  k=60  →  输出 Top-10

④ 重排序 (Cross-Encoder Reranking)
   Top-10 → Cross-Encoder 重打分 → 取 Top-5 进入生成（准确率+8%）

⑤ Context 压缩: 滑动窗口 512 tokens/块，步长128，最终 < 4096 tokens

⑥ 生成: 混元Turbo API → 流式输出

5.3 知识库构建与向量化

企业知识库的构建质量直接决定 RAG 效果：

class DocumentProcessor:
    CHUNK_STRATEGIES = {
        "policy": {"size": 512, "overlap": 128, "strategy": "semantic"},
        "manual":  {"size": 768, "overlap": 64,  "strategy": "recursive"},
        "report":  {"size": 1024,"overlap": 256, "strategy": "fixed"},
    }

    def process(self, file_path: str) -> List[Chunk]:
        ext = Path(file_path).suffix.lower()
        text = self._parse_pdf(file_path) if ext==".pdf" else                self._parse_docx(file_path) if ext==".docx" else                self._parse_markdown(file_path)
        chunks = self._chunk_text(text, self.CHUNK_STRATEGIES.get(self._classify_doc_type(text)))
        chunks = [c for c in chunks if self._is_meaningful(c)]
        embeddings = self.embedding_model.encode([c.text for c in chunks])
        self.vector_store.add(chunks, embeddings)
        return chunks

    def _chunk_text(self, text: str, cfg: dict) -> List[Chunk]:
        paragraphs = text.split("\n\n"); chunks, current = [], ""
        for para in paragraphs:
            if len(current) + len(para) < cfg["size"]:
                current += "\n\n" + para
            else:
                if current: chunks.append(Chunk(text=current.strip(), metadata={}))
                current = para
        if current: chunks.append(Chunk(text=current.strip(), metadata={}))
        return chunks

5.4 流式输出的五大工程挑战

5.5 模型降级与高可用

class ModelRouter:
    MODELS = {
        "primary":   {"name": "混元-turbo",    "timeout": 15, "rate_limit": 100},
        "secondary": {"name": "Qwen-Turbo",    "timeout": 20, "rate_limit": 200},
        "fallback":  {"name": "本地Qwen-7B",   "timeout": 60, "rate_limit": 5},
    }

    async def chat(self, messages: List[dict], model: str = "primary") -> str:
        for attempt in range(2):
            try:
                async with self.semaphore:
                    return await self._call_api(self.MODELS[model], messages)
            except (TimeoutError, RateLimitError) as e:
                logger.warning(f"模型{model}失败: {e}，切换模型"); continue
            except ModelUnavailableError:
                self._mark_model_unavailable(model)
                model = self._get_next_available_model(model); continue
        return await self.chat(messages, model="fallback")

    def _mark_model_unavailable(self, model: str):
        failures = self.redis.incr(f"model:failures:{model}")
        self.redis.expire(f"model:failures:{model}", 300)
        if failures >= 3:
            self.redis.set(f"model:disabled:{model}", "1", ex=300)
            logger.error(f"模型{model}已熔断，5分钟内不再使用")

六、响应聚合层

6.1 统一响应格式

无论走 API 路径还是 LLM 路径，输出格式必须统一：

@dataclass
class UnifiedResponse:
    type: ResponseType      # TEXT | CARD | TABLE | CHART
    content: Any           # 内容体（根据type不同格式）
    source: ResponseSource  # API | LLM | HYBRID
    metadata: ResponseMetadata

class ResponseFormatter:
    @staticmethod
    def format_api_response(api_data: dict, intent: str) -> UnifiedResponse:
        if intent == "查订单":
            return UnifiedResponse(type=ResponseType.TABLE,
                content={"columns":["订单号","商品","金额","状态","时间"],"rows":api_data["orders"]},
                source=ResponseSource.API)
        elif intent == "余额查询":
            return UnifiedResponse(type=ResponseType.CARD,
                content={"title":"账户余额","value":f"¥{api_data['balance']}"},
                source=ResponseSource.API)
        return UnifiedResponse(type=ResponseType.TEXT, content=str(api_data))

    @staticmethod
    def format_llm_response(text: str, sources: List[str]) -> UnifiedResponse:
        return UnifiedResponse(type=ResponseType.TEXT, content=text,
            source=ResponseSource.LLM, metadata=ResponseMetadata(sources=sources))

6.2 Hybrid 混合响应

部分场景需要 API + LLM 协同，比如订单异常分析：用户问"我上个月消费为什么多了300块？" → API 查账单数据 + LLM 生成分析解释，同步返回给用户。

6.3 复杂响应类型渲染策略

七、上下文与状态管理

7.1 Redis 会话状态管理

多轮对话需要在有限上下文窗口内维护会话历史，采用滑动窗口策略管理Token消耗：

class ConversationManager:
    MAX_TOKENS = 128000

    def __init__(self, session_id: str):
        self.session_id = session_id
        self.messages: List[ChatMessage] = self._load_from_redis()

    def add_message(self, role: str, content: str) -> List[ChatMessage]:
        self.messages.append(ChatMessage(role=role, content=content))
        while self.estimate_tokens() > self.MAX_TOKENS * 0.85:
            self.messages.pop(0)
        self._save_to_redis()
        return self.messages

    def estimate_tokens(self) -> int:
        return int(sum(len(m.content) * 1.5 for m in self.messages))

7.2 对话上下文窗口优化策略

• 系统提示词固定：系统提示词固定在窗口底部，不参与滑动淘汰
• RAG Context优先级：上下文满时，优先保留RAG检索结果，裁剪历史对话
• 摘要压缩：超过20轮对话时，自动生成对话摘要替换历史消息
• TTL自动过期：30分钟无交互自动清理Redis释放内存

八、流式输出架构

8.1 SSE与WebSocket的选型

AI对话要求实时流式输出，用户感知"打字中"而非"等待"。对Web端使用SSE，对小程序/APP使用WebSocket，统一底层流式生成器：

@app.post("/api/chat/stream")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        async for token in chat_service.stream_chat(request.session_id, router.route(request.message)):
            yield f"data: {json.dumps({'token': token, 'done': False})}\n\n"
        yield f"data: {json.dumps({'token': '', 'done': True})}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

8.2 Markdown流式渲染的前端处理

• 流式合并：每个Token追加到缓冲区，等待完整语法单元后渲染
• 语法着色：使用highlight.js在渲染线程外预计算，避免主线程阻塞
• Markdown解析：marked.js流式模式，支持增量解析和渲染
• 代码块检测：检测到不完整的代码块时，延迟渲染避免语法错误

九、安全边界设计

9.1 输入安全：敏感信息检测

用户输入可能包含敏感信息（身份证号、银行卡、手机号），需要检测并脱敏后送入LLM：

class SensitiveInfoDetector:
    PATTERNS = {
        "id_card":    r"\d{15}|\d{18}",
        "bank_card": r"\b[1-9]\d{13,19}\b",
        "phone":     r"1[3-9]\d{9}",
        "email":     r"[\w.-]+@[\w.-]+\.\w+",
    }

    def detect_and_mask(self, text: str):
        masked = text; entities = []
        for label, pattern in self.PATTERNS.items():
            for m in re.finditer(pattern, masked):
                s, e = m.span(); display = f"[{label.upper()}]"
                masked = masked[:s] + display + masked[e:]
                entities.append(MaskedEntity(label=label, display=display, start=s))
        return masked, entities

9.2 输出安全：内容安全策略

• 政治/敏感词过滤：关键词库+正则双保险，命中直接拦截返回友好提示
• 幻觉检测：RAG场景下，若检索结果为空，强制LLM输出"未找到相关信息"
• 越权检测：LLM输出中若出现SQL模式，触发安全告警
• 脱敏回显：API返回的用户敏感字段，在聚合层统一脱敏

十、部署与性能优化

10.1 vLLM推理加速

LLM推理是延迟最高的环节，使用vLLM的PagedAttention优化显存管理，吞吐量比naive HF实现高8倍：

from vllm import LLM, SamplingParams
llm = LLM(model="Qwen/Qwen2-72B-Instruct",
          tensor_parallel_size=4, max_model_len=8192,
          gpu_memory_utilization=0.92, enforce_eager=False)
sampling_params = SamplingParams(temperature=0.7, top_p=0.9, max_tokens=1024,
                                  stop=["\n\n", "用户:", "助手:"])

10.2 缓存加速

对于高频、标准化的查询，结果缓存可以完全绕过LLM推理：

class ResponseCache:
    def __init__(self, redis: Redis, ttl: int = 3600):
        self.redis = redis; self.ttl = ttl
    def get(self, key): return self.redis.get(f"response_cache:{key}")
    def set(self, key, val): self.redis.setex(f"response_cache:{key}", self.ttl, val)

10.3 Kubernetes弹性伸缩

• CPU/HPA：CPU>70%时自动扩容，最大10副本
• 自定义指标HPA：基于vllm:pending_requests，LLM请求堆积时快速扩容
• PodDisruptionBudget：保证至少2个Pod存活

十一、测试工程与质量保障

11.1 四层测试体系

AI助手的测试比普通应用复杂得多，需要四层测试体系：

┌─────────────────────────────────────────────────────┐
│  L4 生产监控（灰度+实时告警）                          │
│   - 5%流量灰度，监控路由分布偏移                       │
│   - 新增意图识别准确率实时统计                          │
│   - 用户满意度评分（NPS）实时监控                       │
├─────────────────────────────────────────────────────┤
│  L3 集成测试（端到端）                                 │
│   - 模拟真实用户对话流程，验证完整链路                  │
│   - 前端→网关→路由→API/LLM→聚合→前端                 │
│   - 使用VCR技术录制/回放LLM响应                       │
├─────────────────────────────────────────────────────┤
│  L2 单元测试（模块级）                                 │
│   - 路由引擎：各种输入组合的意图识别准确性              │
│   - RAG：检索结果相关性评分                            │
│   - 响应聚合：不同API数据的格式化正确性                │
│   - 安全过滤：敏感词识别准确率/召回率                   │
├─────────────────────────────────────────────────────┤
│  L1 冒烟测试（PR级别）                                 │
│   - 每条路由路径可通性测试                             │
│   - 核心API端点可用性                                  │
│   - LLM服务健康检查                                    │
└─────────────────────────────────────────────────────┘

11.2 路由引擎评估数据集

我们维护了一个包含5000条标注数据的评估集，覆盖真实用户query分布：

• 高频意图（占60%）：查订单、查余额、退换货、催发货——L1层全覆盖
• 中频意图（占25%）：费用计算、产品推荐、投诉建议——L2层主力
• 长尾意图（占15%）：业务咨询、建议类、分析类——L3层兜底

十二、成果与深度总结

12.1 核心成果指标

12.2 架构设计核心经验

• 双轨不是二选一，而是优先级：API和LLM各有擅长，路由引擎决定谁处理，职责边界必须清晰
• 缓存是性能的天花板：高频标准化查询的缓存命中率直接决定了整体延迟
• RAG的瓶颈在检索而非模型：优化检索策略（RRF+重排序）比换更大参数的模型对准确率提升更显著
• 流式输出的体验设计和技术实现同等重要：Markdown截断等问题不解决，再好的模型输出用户也感知不到
• 可观测性是运维的命脉：全链路打点，任何异常都能快速定位

12.3 后续演进方向

• Function Calling：从意图路由升级为Tool Use，让LLM直接调用API
• 多模态支持：支持图片上传，结合Vision-LLM做图文理解
• 个性化记忆：建立长期用户画像，实现真正的个性化助手体验
• Agent工作流：将AI助手从"回答问题"升级为"执行任务"