LLM模型中迁移学习的实战：用提示实现领域适应

LLM模型中迁移学习的实战：用提示实现领域适应 | 前端进阶之旅

Full translation

完整中文译文

我们要构建一个领域特定的工单分类 Agent，它使用上下文迁移学习来将通用基础模型适配到一个窄分类任务上。与微调权重不同，我们会冻结模型，通过详细的系统 Prompt 和精选的少样本示例来迁移知识。这种方式在 Oxlo.ai 上效果特别好，因为其按请求计费的扁平定价策略让我们可以在每次调用中携带完整上下文示例而不会导致 token 成本膨胀。具体方案详情见定价页。

前置条件

Oxlo.ai 的 API Key（从 https://portal.oxlo.ai 获取）

OpenAI SDK：pip install openai

若干样本工单数据（这里我们用 mock 数据）

步骤 1：定义任务和领域分类体系

首先定义目标分类体系。清晰的分类schema强制模型将其广泛的预训练知识映射到我们特定的领域标签上。我将其放在一个纯 Python 模块中，这样非工程师也能编辑它而无需触碰推理代码。

# schema.py
TAXONOMY = {
    "severity": ["P1-Critical", "P2-High", "P3-Medium", "P4-Low", "P5-Trivial"],
    "component": ["api-gateway", "database", "worker-queue", "frontend", "unknown"]
}

def format_taxonomy() -> str:
    lines = ["Classify each ticket into exactly one severity and one component."]
    lines.append("Severities: " + ", ".join(TAXONOMY["severity"]))
    lines.append("Components: " + ", ".join(TAXONOMY["component"]))
    return "\n".join(lines)

步骤 2：构建迁移 Prompt

系统 Prompt 就是迁移层。它将基础模型的通用分布约束到我们的下游目标上。我将其视为可编辑的配置。你可以调整这段文字来提高准确性，而无需重训练一个权重参数。

SYSTEM_PROMPT = """You are an infrastructure support triage agent.
Your job is to read raw support tickets and classify them using the provided taxonomy.
Respond ONLY with a JSON object containing two keys: "severity" and "component".
Do not add explanations, markdown, or preamble.

Taxonomy:
- severities: P1-Critical, P2-High, P3-Medium, P4-Low, P5-Trivial
- components: api-gateway, database, worker-queue, frontend, unknown

Rules:
- If the ticket mentions connection timeouts or 5xx errors on REST endpoints, use "api-gateway".
- If it mentions disk space, replication lag, or slow queries, use "database".
- If it mentions background job failures or Celery/RabbitMQ, use "worker-queue".
- If it mentions UI rendering or browser errors, use "frontend".
- Default to "unknown" only when no hint is present.
"""

步骤 3：准备少样本示例

接下来构建少样本示例集。这就是不通过梯度下降实现迁移学习的地方。通过向模型展示来自目标领域的输入/输出对，我们将其行为迁移到匹配我们的标注风格。我将示例放在一个列表中，这样可以为不同客户或产品线替换它们。

FEW_SHOT_EXAMPLES = [
    {
        "ticket": "Prod API returning 502 Bad Gateway for /v1/users since 14:03 UTC. Load balancer health checks failing.",
        "output": '{"severity": "P1-Critical", "component": "api-gateway"}'
    },
    {
        "ticket": "Dashboard charts not loading in Chrome. Console shows CORS error on static assets.",
        "output": '{"severity": "P3-Medium", "component": "frontend"}'
    },
    {
        "ticket": "Weekly analytics job stuck for 3 hours. RabbitMQ queue depth at 400k messages.",
        "output": '{"severity": "P2-High", "component": "worker-queue"}'
    },
    {
        "ticket": "Request to increase RDS storage from 500 GB to 1 TB before next growth spike.",
        "output": '{"severity": "P4-Low", "component": "database"}'
    }
]

def build_messages(user_ticket: str) -> list[dict]:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for ex in FEW_SHOT_EXAMPLES:
        messages.append({"role": "user", "content": ex["ticket"]})
        messages.append({"role": "assistant", "content": ex["output"]})
    messages.append({"role": "user", "content": user_ticket})
    return messages

步骤 4：构建 Oxlo.ai 客户端封装

现在将 Prompt 接入推理。Oxlo.ai 暴露了一个 OpenAI 兼容的端点，所以客户端设置只需要改一行代码。我使用 Llama 3.3 70B 作为基础模型，因为它能可靠地遵循结构化指令。扁平化的按请求计费意味着我可以在每次调用中传递完整的系统 Prompt 和所有少样本示例，而无需计算 token 数量。

import json
import os
from openai import OpenAI

client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key=os.environ.get("OXLO_API_KEY"))

def classify_ticket(ticket_text: str) -> dict:
    user_message = ticket_text
    response = client.chat.completions.create(
        model="llama-3.3-70b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.1,
        max_tokens=128,
    )
    raw = response.choices[0].message.content.strip()
    if raw.startswith("```json"):
        raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip()
    return json.loads(raw)

步骤 5：添加批处理和错误处理

生产流水线不会一次只分类一条工单。我将单条工单函数包装在一个带基本错误处理的批处理器中，这样格式错误的 JSON 响应不会导致整个队列崩溃。我还在调用之间添加了小延迟，以便在测试时保持在速率限制之内。

import time
from typing import List

def classify_batch(tickets: List[str], delay_ms: int = 200) -> List[dict]:
    results = []
    for text in tickets:
        try:
            result = classify_ticket(text)
            result["ticket_preview"] = text[:60]
            results.append(result)
        except Exception as exc:
            results.append({
                "ticket_preview": text[:60],
                "error": str(exc),
                "severity": "unknown",
                "component": "unknown"
            })
        time.sleep(delay_ms / 1000.0)
    return results

我使用三条未见过的工单测试流水线。前两条应该能干净地映射到我们的分类体系，而第三条故意写得模糊以测试回退行为。

if __name__ == "__main__":
    unseen_tickets = [
        "Payment webhook worker crashing with SIGKILL after processing 12k events. OOM on queue consumer pod.",
        "Users reporting 404 on /billing/invoices after latest deploy. Nginx ingress routing table looks correct.",
        "The office plants need watering and the coffee machine is out of beans."
    ]

    for row in classify_batch(unseen_tickets):
        print(row)

当我用 Oxlo.ai 运行这个时，输出如下：

{'severity': 'P2-High', 'component': 'worker-queue', 'ticket_preview': 'Payment webhook worker crashing with SIGKILL ...'}
{'severity': 'P2-High', 'component': 'api-gateway', 'ticket_preview': 'Users reporting 404 on /billing/invoices aft...'}
{'severity': 'P5-Trivial', 'component': 'unknown', 'ticket_preview': 'The office plants need watering and the coff...'}

将静态少样本示例替换为通过 Oxlo.ai 的 BGE-Large embeddings 端点动态检索到的示例。将历史工单存储在向量数据库中，嵌入传入的工单，然后拉取 top 3 最近邻作为上下文。这将基于 Prompt 的迁移学习转变为检索增强流水线，随着工单量增长自动改进。

使用相同 Prompt 和评估集在 Oxlo.ai 上的不同基础模型之间运行 A/B 测试，例如 Qwen 3 32B 或 Kimi K2.6。不同架构以不同速率将知识迁移到专业领域，而按请求计费使并行比较它们变得便宜，无需配置单独的基础设施。