LLM 统一网关设计 —— LiteLLM 抽象、Prompt 工程与响应清洗 · devcfg

🎯 本章核心问题

在 AI 应用开发中，如何优雅地调用 LLM API？ 直接用 requests.post() 硬编码？还是每个业务模块都写一套调用逻辑？

我们面临的具体挑战：

• ❌ 模型锁定风险：今天用 DeepSeek，明天想换 GPT-4，要改几十处代码
• ❌ 输出不稳定：LLM 偶尔会返回 ````sql 包裹的代码、` 标签、或者多余的解释文字
• ❌ Token 超限崩溃：语义模型太大时直接报错 context_length_exceeded
• ❌ 调试困难：不知道 LLM 收到了什么 Prompt、返回了什么、耗时多少

解决方案：构建一个统一的 LLM Gateway（网关层），将所有 LLM 交互收敛到一个模块。

---

🏗️ 架构总览

LLM 统一网关架构

设计原则

单一职责、开闭原则、可观测性与防御性编程

---

🔌 一、LiteLLM 统一抽象层

为什么不直接用 OpenAI SDK？

对比维度	OpenAI SDK	LiteLLM
多模型支持	❌ 仅 OpenAI	✅ DeepSeek/OpenAI/Claude/Gemini 等 100+
接口统一	❌ 各家参数不同	✅ 统一 acompletion() 接口
错误标准化	❌ 各家错误码不同	✅ 统一异常类型
重试机制	❌ 需自己实现	✅ 内置指数退避重试
成本追踪	❌ 无	✅ 内置 Token 用量统计

核心实现

app/services/llm_gateway.py

import litellm

class LLMGateway:
    def __init__(self):
        self.provider = settings.LLM_PROVIDER    # "deepseek"
        self.model = settings.LLM_MODEL          # "deepseek-chat"
        self.api_key = settings.LLM_API_KEY
        self.api_base = settings.LLM_API_BASE    # "https://api.deepseek.com"

    def _get_model_name(self, model=None):
        """格式化模型名称以适配 LiteLLM"""
        if model:
            return model
        if "deepseek" in self.model.lower():
            return f"deepseek/{self.model}"   # → "deepseek/deepseek-chat"
        return f"openai/{self.model}"         # → "openai/gpt-4o"

💡 关键设计：模型名格式化

LiteLLM 要求使用 provider/model 格式：

# DeepSeek 格式
"deepseek/deepseek-chat"       # 对应 DeepSeek-V3
"deepseek/deepseek-coder"      # 对应 DeepSeek-Coder

# OpenAI 格式
"openai/gpt-4o"
"openai/gpt-4-turbo"

# 自定义端点（兼容 OpenAI API 的私有部署）
"openai/my-custom-model"       # 配合 api_base 使用

切换模型的成本 = 修改 1 行 .env 配置 🎉

---

📝 二、Prompt 工程实践

2.1 System Prompt 模板库

我们的系统有 4 种不同的 LLM 调用场景，每种都需要精心设计的 System Prompt：

场景 1：SQL 生成（最复杂）

app/services/llm_gateway.py

async def generate_sql(self, natural_language, semantic_model):
    system_prompt = """你是一个专业的 SQL 生成专家。根据用户的自然语言查询和数据库语义模型，
生成合法的 MySQL SQL 查询语句。

规则：
1. 只输出 SQL 语句，不要输出任何解释文字。
2. 使用语义模型中正确的表名和字段名。
3. 需要关联查询时，根据表关系使用 JOIN。
4. 根据需要添加 WHERE、GROUP BY、ORDER BY 等子句。
5. 如果用户没有明确指定排序，默认按主键或时间字段降序排列。"""

    user_prompt = f"""数据库语义模型：
{json.dumps(semantic_model, ensure_ascii=False, indent=2)}

用户的自然语言查询：{natural_language}

请生成 SQL："""

设计要点：

• ✅ 约束性指令：”只输出 SQL”、”不要解释” → 减少噪声
• ✅ 上下文注入：将完整的语义模型作为上下文传入
• ✅ 规则明确：排序、JOIN 等边界情况给出默认策略

场景 2：元数据分析（结构化输出）

app/services/llm_gateway.py

async def analyze_metadata(self, tables_info):
    system_prompt = """你是一位资深的数据库分析师。请分析给定的数据库表结构，
完成以下任务：
1. 为每张表起一个通俗易懂的中文名称（业务名称）
2. 为每张表写一段中文业务描述
3. 为每个字段起一个中文业务名称
...

请返回如下 JSON 结构：
{
  "tables": {
    "表名": {
      "business_name": "中文业务名称",
      ...
    }
  },
  "suggested_relations": [
    {
      "from_table": "源表名",
      "relation_type": "one_to_one 或 one_to_many",
      "confidence": "high 或 medium 或 low",
      ...
    }
  ]
}

重要：只输出 JSON，不要输出任何其他文字。"""

设计要点：

• ✅ 强制 JSON 输出：便于程序解析
• ✅ 置信度字段：标记关系推断的可信度（high/medium/low）
• ✅ 示例驱动：通过完整的 JSON 模板引导输出格式

2.2 Prompt 注入技巧

变量模板化

# ❌ 错误做法：字符串拼接容易出错
prompt = "数据库有这些表：" + str(tables) + "用户问：" + question

# ✅ 正确做法：f-string + json.dumps 保证格式
user_prompt = f"""数据库语义模型：
{json.dumps(semantic_model, ensure_ascii=False, indent=2)}  # 缩进美化

用户的自然语言查询：{natural_language}

请生成 SQL："""

Few-shot 示例（可选增强）

对于复杂任务，可以在 User Prompt 中追加示例：

user_prompt += """

示例：
用户输入："查询销售额最高的前10个商品"
输出：
SELECT p.name, SUM(oi.amount) as total_sales
FROM products p
JOIN order_items oi ON p.id = oi.product_id
GROUP BY p.id
ORDER BY total_sales DESC
LIMIT 10;

现在请根据上述规则生成 SQL："""

---

🔒 三、长度控制机制

问题背景

当数据库有很多表（50+）且字段很多（200+）时，语义模型 JSON 可能超过 80K 字符，直接导致 LLM API 报错：

Error: This model's maximum context length is 65536 tokens...

解决方案：智能截断

app/services/llm_gateway.py

async def generate(self, prompt, system_prompt=None, model=None):
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": prompt})

    # ⚡ 关键：长度检查与截断
    total_chars = sum(len(m["content"]) for m in messages)
    if total_chars > 80000:
        logger.warning(f"[LLM Request] prompt too long ({total_chars} chars), truncating")
        for m in messages:
            if len(m["content"]) > 60000:
                m["content"] = m["content"][:60000] + "\n...（内容过长已截断）"

    # 调用 LLM
    response = await litellm.acompletion(
        model=self._get_model_name(model),
        messages=messages,
        api_key=self.api_key,
        api_base=self.api_base,
    )
    return response.choices[0].message.content

截断策略说明

参数	阈值	说明
总字符数上限	80,000	约等于 20K tokens（保守估计）
单条消息上限	60,000	防止单条消息占满所有空间
截断位置	60,000 字符处	保留前半部分（通常包含最重要的信息）
截断提示	追加 “…（内容过长已截断）”	让 LLM 知道数据不完整

💡 为什么选择截断而不是压缩？

方案	优点	缺点
截断（当前方案）	实现简单；保留原始数据格式	可能丢失尾部重要信息
摘要压缩（LLM 先总结）	保留关键信息	增加一次 LLM 调用；延迟翻倍
分块多次调用	不丢失数据	复杂度高；需要合并结果

我们的选择：对于大多数场景（<30 张表），不会触发截断；即使触发，前 60K 字符通常已覆盖核心表结构。

---

🧹 四、响应清洗器

问题背景

不同 LLM 的输出风格差异很大：

LLM	典型输出示例
DeepSeek-V3	\n\n<think >\n用户想查询...\n</think >\n\n\``sql\nSELECT …\n```\n\n当然，您也可以…`
GPT-4o	Here's the SQL query:\n\``sql\nSELECT …\n```\n\nThis query will…`
Claude	Certainly! Here's the query:\n\``\nSELECT …\n```\n\nLet me explain…`

共同特点：都包含噪声（思考过程、代码块标记、解释文字）

清洗流程

app/services/llm_gateway.py (generate_sql 方法)

result = await self.generate(user_prompt, system_prompt=system_prompt)
result = result.strip()

# 步骤 1：去除 <think > 标签（DeepSeek 特有）
think_match = re.search(r'</think\s*>', result, re.DOTALL)
if think_match:
    result = result[think_match.end():].strip()  # 保留 </think > 之后的内容

# 步骤 2：去除 ```sql 开头标记
if result.startswith("```sql"):
    result = result[6:]

# 步骤 3：去除 ``` 开头标记（通用代码块）
if result.startswith("```"):
    result = result[3:]

# 步骤 4：去除 ``` 结尾标记
if result.endswith("```"):
    result = result[:-3]

return result.strip()

对于 JSON 输出的额外清洗

app/services/llm_gateway.py (analyze_metadata 方法)

# 提取 JSON（可能被包裹在 ```json 中）
json_match = re.search(r'\{[\s\S]*\}', result)
if json_match:
    result = json_match.group(0)

# 剥离代码块标记
if result.startswith("```json"):
    result = result[7:]
if result.startswith("```"):
    result = result[3:]
if result.endswith("```"):
    result = result[:-3]

# 尝试解析并校验
try:
    parsed = json.loads(result.strip())
    logger.info(f"[analyze_metadata] parsed successfully, tables={len(parsed.get('tables', {}))}")
    return parsed
except json.JSONDecodeError as e:
    logger.error(f"[analyze_metadata] JSON parse failed: {e}")
    return {}  # 返回空字典而非抛异常，保证系统稳定

---

📊 五、可观测性：日志系统

日志记录策略

app/services/llm_gateway.py

logger = logging.getLogger("llm_gateway")

async def generate(self, prompt, system_prompt=None, ...):
    # 记录请求信息
    logger.info(f"[LLM Request] model={kwargs['model']}, "
                f"api_base={self.api_base}, "
                f"total_chars={total_chars}")
    logger.info(f"[LLM Request] system_prompt (first 500):\n{(system_prompt or '')[:500]}")
    logger.info(f"[LLM Request] user_prompt (first 1000 chars):\n{prompt[:1000]}")

    start_time = time.time()
    try:
        response = await litellm.acompletion(**kwargs)
        elapsed = time.time() - start_time
        content = response.choices[0].message.content

        # 记录响应信息
        logger.info(f"[LLM Response] elapsed={elapsed:.2f}s, length={len(content)}")
        logger.info(f"[LLM Response Content] (first 2000 chars):\n{content[:2000]}")
        return content
    except Exception as e:
        elapsed = time.time() - start_time
        logger.error(f"[LLM Error] elapsed={elapsed:.2f}s, "
                     f"error={type(e).__name__}: {e}")
        raise

日志输出示例

INFO     [LLM Request] model=deepseek/deepseek-chat, api_base=https://api.deepseek.com, total_chars=45230
INFO     [LLM Request] system_prompt (first 500):
你是一个专业的 SQL 生成专家。根据用户的自然语言查询和数据库语义模型，
生成合法的 MySQL SQL 查询语句。
规则：
1. 只输出 SQL 语句，不要输出任何解释文字。
...
INFO     [LLM Request] user_prompt (first 1000 chars):
数据库语义模型：
{
  "tables": [
    {
      "table_name": "orders",
      "fields": [...]
    }
  ]
}
用户的自然语言查询：查询最近7天的订单总额
...
INFO     [LLM Response] elapsed=2.34s, length=187
INFO     [LLM Response Content] (first 2000 chars):
```sql
SELECT DATE(created_at) as date, SUM(total_amount) as total
FROM orders
WHERE created_at >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
GROUP BY DATE(created_at)
ORDER BY date DESC;

#### 💡 日志的价值

| 场景 | 如何利用日志 |
|------|-------------|
| **调试 Prompt 质量** | 查看 `user_prompt` 确认语义模型是否正确注入 |
| **性能优化** | 通过 `elapsed` 定位慢请求（通常 >5s 需要优化） |
| **成本控制** | 统计 `total_chars` 估算 Token 用量 |
| **错误排查** | 查看完整的 `[LLM Error]` 堆栈信息 |

---

## 🔧 六、统一接口设计

### Gateway 对外暴露的方法

| 方法名 | 调用方 | 输入 | 输出 | 用途 |
|--------|--------|------|------|------|
| `generate(prompt, system_prompt)` | Chat Engine | 文本 | 文本 | 通用对话 |
| `chat_completion(messages)` | Chat Engine | 消息数组 | 文本 | 多轮对话 |
| `generate_sql(nl, semantic_model)` | SQL Generator | NL + 模型 | SQL 字符串 | NL→SQL |
| `analyze_metadata(tables_info)` | Metadata Service | 表结构列表 | JSON | 元数据分析 |
| `generate_dashboard_config(desc, model)` | Dashboard Engine | 描述 + 模型 | JSON 数组 | 大屏配置 |
| `generate_report_script(data, metrics)` | Report Generator | 数据 + 指标 | Markdown | 报告生成 |

### 调用示例

#### 示例 1：SQL Generator 调用

```python
# sql_generator.py
from app.services.llm_gateway import LLMGateway

async def generate_query(natural_language, semantic_model):
    gateway = LLMGateway()
    sql = await gateway.generate_sql(natural_language, semantic_model)
    return sql

generate_sql → generate → litellm.acompletion → 清洗后的 SQL

示例 2：Chat Engine 调用

# chat_engine.py
gateway = LLMGateway()
assistant_content = await gateway.chat_completion([
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": conversation_text},
])

---

⚠️ 七、异常处理策略

异常分类与处理

try:
    response = await litellm.acompletion(**kwargs)
except litellm.exceptions.AuthenticationError as e:
    # API Key 无效
    logger.error(f"[LLM] Authentication failed: {e}")
    raise ValueError("LLM API Key 无效，请检查配置")
except litellm.exceptions.RateLimitError as e:
    # 请求频率超限
    logger.warning(f"[LLM] Rate limited: {e}")
    raise Exception("LLM 请求过于频繁，请稍后重试")
except litellm.exceptions.ContextWindowExceededError as e:
    # 上下文超长（虽然我们有长度控制，但保险起见）
    logger.error(f"[LLM] Context window exceeded: {e}")
    raise Exception("输入内容过长，请简化查询条件")
except Exception as e:
    # 其他未知错误
    logger.error(f"[LLM] Unknown error: {type(e).__name__}: {e}")
    raise Exception(f"LLM 服务异常: {str(e)}")

💡 防御性编程原则

1. 永不信任 LLM 输出：始终做清洗和校验
2. 优雅降级：返回空值而非崩溃（如 JSON 解析失败返回 {}）
3. 详细日志：记录足够的信息用于事后分析
4. 超时控制：虽然 LiteLLM 有内置超时，但建议业务层也设置超时

---

📈 八、性能优化技巧

8.1 并发调用

当大屏有多个 Widget 需要解析时，可以并发调用 LLM：

import asyncio

async def resolve_dashboard_queries(db, dashboard_id):
    tasks = []
    for widget in dashboard.widgets:
        task = asyncio.create_task(
            resolve_nl_to_sql(db, widget.data_source_id, widget.nl_query)
        )
        tasks.append(task)

    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 10 个 Widget 并发解析：10s → ~3s（假设每个 2-3s）

8.2 缓存语义模型

语义模型（表结构+关系）相对稳定，可以缓存避免重复传输：

from functools import lru_cache
import hashlib

@lru_cache(maxsize=10)
def get_semantic_model_cache(datasource_id: int, hash_value: str):
    """缓存语义模型（基于 datasource_id + 内容哈希）"""
    return _build_semantic_model(datasource_id)

def cache_key(datasource_id, model_dict):
    return (datasource_id, hashlib.md5(json.dumps(model_dict).encode()).hexdigest())

8.3 流式输出（未来优化方向）

当前使用同步等待模式，未来可改为流式输出提升用户体验：

# 伪代码：流式响应
async def stream_generate(prompt):
    async for chunk in litellm.acompletion_stream(...):
        yield chunk.choices[0].delta.content  # 逐 token 返回

---

🎯 九、实战案例演示

案例：从自然语言到可执行 SQL

输入：

用户查询："查询上个月每个品类的销售额排名"

Step 1 - 构建 Prompt：

{
  "system": "你是一个专业的 SQL 生成专家...",
  "user": "数据库语义模型：{...45000字的JSON...}\n\n用户查询：..."
}

Step 2 - 长度检查：

✅ 总字符数：45,230 (< 80,000) → 无需截断

Step 3 - 调用 LLM：

⏱️ 耗时：2.34s
📍 模型：deepseek/deepseek-chat
💰 成本：约 ¥0.023（基于 2K tokens）

Step 4 - 原始响应：

\n\n<think >
用户想查询"上个月每个品类的销售额排名"，需要：
1. 筛选时间范围（上个月）
2. 按品类分组
3. 计算销售额总和
4. 按销售额降序排列
涉及表：orders, order_items, products, categories
...</think >

```sql
SELECT c.name AS category_name,
       SUM(oi.amount) AS total_sales
FROM categories c
JOIN products p ON c.id = p.category_id
JOIN order_items oi ON p.id = oi.product_id
JOIN orders o ON o.id = oi.order_id
WHERE o.created_at >= '2026-04-01'
  AND o.created_at < '2026-05-01'
GROUP BY c.id, c.name
ORDER BY total_sales DESC;

**Step 5 - 清洗后结果**：
```sql
SELECT c.name AS category_name,
       SUM(oi.amount) AS total_sales
FROM categories c
JOIN products p ON c.id = p.category_id
JOIN order_items oi ON p.id = oi.product_id
JOIN orders o ON o.id = oi.order_id
WHERE o.created_at >= '2026-04-01'
  AND o.created_at < '2026-05-01'
GROUP BY c.id, c.name
ORDER BY total_sales DESC;

完美！ 🎉 可直接交给 execute_query() 执行。

---

🔟 十、总结与最佳实践

✅ 我们做到了什么

1. 零成本模型切换：一行配置从 DeepSeek 切到 GPT-4
2. 稳定的输出质量：多层清洗确保返回干净的 SQL/JSON
3. 完善的可观测性：日志帮助快速定位问题
4. 防御性编程：长度控制 + 异常处理保证稳定性

📚 最佳实践清单

• 使用 LiteLLM 统一多模型调用
• System Prompt 明确约束输出格式
• 实现 Prompt 长度检查与截断
• 编写响应清洗器（去标签、去代码块标记）
• 记录详细的请求/响应日志
• 区分不同场景的异常类型
• 考虑并发调用提升吞吐量
• 缓存相对稳定的上下文（如语义模型）

🚀 下一步

下一篇文章我们将深入 元数据管理系统 —— 如何让 LLM 自动理解数据库结构？如何智能推断表关系？如何构建高质量的语义模型？

敬请期待！🚀

---

相关代码文件：app/services/llm_gateway.py（llm_gateway.py）
配置文件：app/config.py（config.py）、.env.example（.env.example）
依赖库：litellm 文档