“

系列：S13 AI 工程 · 第 8 篇 | 难度：高级 | 阅读时间：18 分钟

LLM 编排与路由策略

#TL;DR

企业通常同时使用多个 LLM（GPT-4、Claude、本地部署的开源模型），如何根据任务复杂度、成本预算和延迟要求智能路由到最合适的模型？本文介绍 coomia-dip 的 LLM Router 设计，包括基于规则的路由、成本感知路由、级联降级和模型性能监控。

#1. 背景与挑战

#1.1 企业 AI 的现实困境

在企业级 AI 工程实践中，技术选型只是起点。真正的挑战在于如何将 AI 能力无缝融入已有的数据平台和业务流程。coomia-dip 作为本体驱动的智能决策平台，天然具备将结构化知识（Ontology）与非结构化 AI 能力（LLM/向量检索）融合的架构优势。

本文聚焦的核心问题是：企业通常同时使用多个 LLM（GPT-4、Claude、本地部署的开源模型），如何根据任务复杂度、成本预算和延迟要求智能路由到最合适的模型？本文介绍 coomia-dip 的 LLM Router 设计，包括基于规则的路由、成本感知路由、级联降级和模型性能监控。

#1.2 行业现状

当前企业 AI 落地面临几个共性问题：

碎片化：AI 能力散落在不同的系统和团队中，缺乏统一治理
可观测性差：AI 系统的决策过程不透明，难以排查问题
成本失控：LLM API 调用成本随规模线性增长，缺乏优化手段
安全合规：企业数据通过 AI 系统流转，安全边界模糊

coomia-dip 的 AI 工程体系旨在系统性解决这些问题。

#2. 架构设计

#2.1 整体架构

在 coomia-dip 的八平面架构中，AI 能力主要分布在 Reasoning & Decision Layer（Reasoning & Decision）和 Agent Runtime Layer（Agent Runtime）：

Python

# 核心组件
class AIArchitecture:
    """coomia-dip AI 工程架构"""

    # Reasoning & Decision Layer: 推理与决策
    reasoning_engine: ReasoningEngine      # 规则引擎 + 推理
    vector_store: VectorStore              # 向量存储（Doris HNSW）
    embedding_pipeline: EmbeddingPipeline  # 向量化管道
    llm_router: LLMRouter                  # LLM 路由与编排

    # Agent Runtime Layer: Agent 运行时
    agent_runtime: AgentRuntime            # Agent 执行环境
    tool_registry: ToolRegistry            # 工具注册中心
    workflow_engine: TemporalClient        # 工作流引擎
    memory_store: MemoryStore              # 对话记忆存储

#2.2 核心数据流

Code

用户输入 -> 意图识别 -> 路由决策
    |                       |
    |   +-------------------+-------------------+
    |   |                   |                   |
    v   v                   v                   v
  OQL 查询         LLM 推理           工作流触发
    |                   |                   |
    v                   v                   v
  Ontology          向量检索             Temporal
  数据层          (RAG Context)         工作流引擎
    |                   |                   |
    +-------------------+-------------------+
                        |
                        v
                    结果融合
                        |
                        v
                    输出格式化
                        |
                        v
                    安全过滤
                        |
                        v
                    用户响应

#3. 核心实现

#3.1 配置与初始化

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.ai import AIConfig

platform = OntoPlatform(
    base_url="http://localhost:8080",
    token="admin-token",
)

# AI 功能配置
ai_config = AIConfig(
    embedding_model="text-embedding-3-small",
    embedding_dimension=768,
    llm_models={
        "default": {
            "provider": "openai",
            "model": "gpt-4o",
            "temperature": 0.1,
            "max_tokens": 4096,
        },
        "fast": {
            "provider": "openai",
            "model": "gpt-4o-mini",
            "temperature": 0.0,
            "max_tokens": 2048,
        },
        "local": {
            "provider": "ollama",
            "model": "llama3:8b",
            "endpoint": "http://ollama:11434",
        },
    },
    vector_store={
        "type": "doris",
        "index_type": "hnsw",
        "metric": "cosine",
    },
    safety={
        "pii_detection": True,
        "content_filter": True,
        "max_input_tokens": 8192,
        "max_output_tokens": 4096,
    },
)

platform.ai.configure(ai_config)

#3.2 核心功能实现

Python

# 功能实现示例
from ontology_sdk.ai import (
    EmbeddingPipeline,
    SemanticSearch,
    LLMChain,
    SafetyFilter,
)

# 1. 向量化管道
pipeline = EmbeddingPipeline(
    model=ai_config.embedding_model,
    batch_size=100,
    dimension=ai_config.embedding_dimension,
)

# 处理文档
documents = platform.objects.list("Document", limit=1000)
for doc in documents:
    chunks = pipeline.chunk(doc.content, chunk_size=512, overlap=50)
    embeddings = pipeline.embed(chunks)
    for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
        platform.objects.create("KnowledgeChunk", {
            "documentId": doc.documentId,
            "chunkIndex": i,
            "content": chunk,
            "embedding": embedding,
        })

# 2. 语义搜索
search = SemanticSearch(platform)
results = search.query(
    query="如何配置多租户隔离?",
    object_type="KnowledgeChunk",
    top_k=5,
    filters={"category": "configuration"},
)

# 3. RAG 链
rag_chain = LLMChain(
    model="default",
    system_prompt="""你是 coomia-dip 平台的技术助手。
基于提供的上下文回答用户问题。如果上下文中没有相关信息，请明确说明。""",
    context_builder=lambda query: search.query(query, top_k=5),
    output_parser="markdown",
)

response = rag_chain.invoke("如何配置多租户隔离?")
print(response.content)
print(f"Sources: {[r.source for r in response.sources]}")

# 4. 安全过滤
safety = SafetyFilter(ai_config.safety)
filtered_input = safety.filter_input(user_input)
filtered_output = safety.filter_output(llm_response)

#4. 高级应用场景

#4.1 场景一：智能数据探索

Python

# 用户通过自然语言查询 Ontology 数据
async def natural_language_query(user_question: str) -> dict:
    # Step 1: 将自然语言转换为 OQL
    oql = await platform.ai.text_to_oql(
        question=user_question,
        context_object_types=["Order", "Customer", "Product"],
    )
    print(f"Generated OQL: {oql}")

    # Step 2: 执行查询
    result = platform.oql.execute(oql)

    # Step 3: 生成自然语言摘要
    summary = await platform.ai.summarize(
        data=result.rows,
        question=user_question,
        format="bullet_points",
    )

    return {
        "oql": oql,
        "data": result.rows,
        "summary": summary,
    }

#4.2 场景二：异常检测与告警

Python

# 基于 AI 的时序异常检测
async def detect_anomalies(metric_name: str, window_hours: int = 24):
    # 获取历史数据
    data = platform.metrics.query(
        metric=metric_name,
        duration=f"{window_hours}h",
        granularity="5m",
    )

    # AI 分析异常
    analysis = await platform.ai.analyze_timeseries(
        data=data.values,
        timestamps=data.timestamps,
        sensitivity="medium",
    )

    if analysis.anomalies:
        for anomaly in analysis.anomalies:
            print(f"Anomaly at {anomaly.timestamp}: {anomaly.description}")
            print(f"  Severity: {anomaly.severity}")
            print(f"  Root cause hypothesis: {anomaly.hypothesis}")

    return analysis

#4.3 场景三：知识图谱增强

Python

# 利用 LLM 从非结构化文档中提取实体和关系
async def extract_knowledge(document: str) -> dict:
    extraction = await platform.ai.extract_entities(
        text=document,
        entity_types=["Person", "Organization", "Technology", "Concept"],
        relation_types=["worksAt", "uses", "relatedTo"],
        ontology_context=platform.ontology.get_schema(),
    )

    # 自动写入 Ontology
    for entity in extraction.entities:
        platform.objects.create(entity.type, entity.properties)

    for relation in extraction.relations:
        platform.links.create(
            relation.source_rid,
            relation.relation_type,
            relation.target_rid,
        )

    return extraction

#5. 性能优化

#5.1 关键优化点

优化方向	技术手段	预期效果
向量检索延迟	HNSW 参数调优 + 预热	P99 < 50ms
LLM 调用成本	模型路由 + 缓存	成本降低 40-60%
Embedding 吞吐	批量处理 + GPU 加速	10x 提升
端到端延迟	流式输出 + 并行检索	首字延迟 < 500ms

#5.2 缓存策略

Python

# 语义缓存：相似问题直接返回缓存结果
class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.threshold = similarity_threshold

    async def get(self, query: str) -> str | None:
        query_embedding = await embed(query)
        similar = await vector_search(query_embedding, top_k=1)
        if similar and similar[0].score >= self.threshold:
            return similar[0].cached_response
        return None

    async def set(self, query: str, response: str):
        query_embedding = await embed(query)
        await store(query_embedding, response)

#6. 监控与可观测性

#6.1 AI 操作指标

Python

# 关键监控指标
ai_metrics = platform.ai.get_metrics(duration="1h")

print(f"=== LLM 指标 ===")
print(f"  Total calls: {ai_metrics.llm.total_calls}")
print(f"  Avg latency: {ai_metrics.llm.avg_latency_ms}ms")
print(f"  Total tokens: {ai_metrics.llm.total_tokens}")
print(f"  Estimated cost: ${ai_metrics.llm.estimated_cost:.2f}")

print(f"\n=== 向量检索指标 ===")
print(f"  Total queries: {ai_metrics.vector.total_queries}")
print(f"  Avg latency: {ai_metrics.vector.avg_latency_ms}ms")
print(f"  Cache hit rate: {ai_metrics.vector.cache_hit_rate:.1%}")

print(f"\n=== 安全指标 ===")
print(f"  PII detections: {ai_metrics.safety.pii_detections}")
print(f"  Content filter blocks: {ai_metrics.safety.filter_blocks}")

#7. 最佳实践与经验教训

#7.1 设计原则

AI 是增强而非替代：AI 增强 Ontology 操作的效率，但关键决策仍需人类确认
可观测性优先：每个 AI 调用都应被记录和可追踪
渐进式采用：从低风险场景（搜索增强）开始，逐步扩展到高风险场景（决策辅助）
成本意识：监控 Token 消耗，为每个 AI 功能设置预算上限
安全兜底：所有 AI 输出都必须经过安全过滤，不直接暴露给终端系统

#7.2 常见陷阱

陷阱	后果	解决方案
盲目使用最大模型	成本爆炸	按任务复杂度路由到合适的模型
忽略 Prompt 版本管理	难以复现问题	Prompt 纳入版本控制
未限制输出长度	Token 浪费	设置 max_tokens 和输出格式约束
未做 PII 过滤	数据泄露风险	输入/输出双向 PII 检测
缓存粒度过粗	缓存命中率低	语义缓存 + 参数化查询缓存

#总结

本文深入探讨了LLM 编排与路由策略在 coomia-dip 中的设计与实现。核心要点包括：架构设计原则、关键功能实现、高级应用场景、性能优化策略、监控可观测性和实践经验总结。AI 工程不是单点技术的堆砌，而是需要平台级的系统性设计——coomia-dip 提供了这样的基础设施。

下一篇：[S13-09] {next_title} 上一篇：[S13-07] {prev_title}