回归测试怎么做：用失败样本库驱动提示词路由工具持续迭代

副标题： 构建高可靠LLM应用的质量保障体系——从理论到实践的完整指南

---

摘要/引言

在大语言模型（LLM）应用迅速普及的今天，如何保证LLM应用的稳定性和可靠性成为了开发者面临的核心挑战。传统的软件测试方法在面对LLM的非确定性输出时显得力不从心，而回归测试更是成为了一大难题。

本文将深入探讨一种创新的回归测试方法：通过构建失败样本库，驱动提示词路由工具的持续迭代。我们将从问题背景出发，逐步介绍核心概念、系统设计、实现方案以及最佳实践，帮助读者构建一套完整的LLM应用质量保障体系。

读完本文，你将：

• 理解LLM应用回归测试的特殊性与挑战
• 掌握失败样本库的设计与构建方法
• 学会如何设计和实现提示词路由工具
• 了解如何通过失败样本驱动系统持续迭代
• 获取完整的代码实现和最佳实践指南

---

目标读者与前置知识

目标读者：

• 有一定AI应用开发经验的开发者
• 对大语言模型应用有兴趣的软件工程师
• 负责AI产品质量保证的测试工程师
• 希望优化LLM应用性能的技术团队负责人

前置知识：

• 基本的Python编程能力
• 对大语言模型API（如OpenAI GPT、Claude等）有基本了解
• 熟悉软件测试的基本概念
• 了解版本控制（Git）的基本使用

---

文章目录

1. 引言与基础

   • 摘要/引言
   • 目标读者与前置知识
   • 文章目录

2. 问题背景与动机

   • LLM应用的崛起与挑战
   • 传统测试方法的局限性
   • 为什么回归测试在LLM应用中如此困难
   • 失败样本库与提示词路由的价值

3. 核心概念与理论基础

   • LLM应用回归测试的核心概念
   • 失败样本库的设计原理
   • 提示词路由的基本原理
   • 持续迭代的反馈循环机制

4. 环境准备

   • 技术栈选择
   • 开发环境配置
   • 依赖库安装
   • 项目结构设计

5. 分步实现：失败样本库

   • 样本数据模型设计
   • 数据采集与标注流程
   • 样本库存储方案
   • 样本库管理功能实现

6. 分步实现：提示词路由工具

   • 路由策略设计
   • 提示词模板管理
   • 路由引擎实现
   • 集成与测试

7. 关键代码解析与深度剖析

   • 样本相似度计算算法
   • 路由决策逻辑
   • 反馈循环机制
   • 性能优化考量

8. 结果展示与验证

   • 系统功能演示
   • 测试结果分析
   • 性能指标对比

9. 性能优化与最佳实践

   • 样本库管理优化
   • 路由效率提升
   • 团队协作流程
   • 持续监控与改进

10. 常见问题与解决方案

    • 样本质量问题
    • 路由决策不准确
    • 系统性能瓶颈
    • 团队协作挑战

11. 未来展望与扩展方向

    • 技术发展趋势
    • 可能的改进方向
    • 行业应用前景

12. 总结

13. 参考资料

14. 附录

---

问题背景与动机

LLM应用的崛起与挑战

近年来，大语言模型（LLM）技术取得了突破性进展，从GPT-3到GPT-4，从Claude到PaLM，模型的能力不断提升。随之而来的是LLM应用的爆发式增长，从智能客服到代码助手，从内容生成到数据分析，LLM正在深刻改变着我们构建软件的方式。

然而，随着LLM应用的普及，开发者们也面临着前所未有的挑战：

1. 输出非确定性：相同的输入可能产生不同的输出，这使得传统的断言式测试难以应用。
2. 行为不可预测：模型可能在某些边缘案例上表现出意外的行为。
3. 版本更新风险：模型提供商更新底层模型时，可能会破坏现有应用的功能。
4. 提示词脆弱性：细微的提示词变化可能导致输出质量的大幅波动。

传统测试方法的局限性

在传统软件开发中，我们有一套成熟的测试方法论：单元测试、集成测试、端到端测试等。这些方法依赖于可预测的输入输出行为，但在LLM应用中，这些假设往往不成立。

例如，考虑一个简单的LLM应用：一个文本摘要工具。对于相同的输入文本，模型可能生成略有不同但同样合理的摘要。传统的断言测试会因为这种差异而失败，但实际上应用是正常工作的。

此外，LLM应用的"正确性"往往是主观的，取决于用户的期望和上下文。这使得定义明确的测试标准变得异常困难。

为什么回归测试在LLM应用中如此困难

回归测试是保证软件质量的关键环节，它确保在添加新功能或修改现有代码时，不会破坏已有的功能。但在LLM应用中，回归测试面临着特殊的挑战：

1. 模型更新的影响：当底层LLM模型更新时，即使你的应用代码没有变化，行为也可能发生改变。
2. 提示词调整的风险：优化提示词是提升LLM应用性能的常用手段，但任何提示词的修改都可能在某些场景下引入新问题。
3. 评估成本高昂：人工评估LLM输出的质量既费时又昂贵，难以在每次变更后进行全面的回归测试。
4. 缺乏标准化的测试框架：虽然社区正在探索各种解决方案，但目前还没有像JUnit或pytest那样被广泛接受的LLM测试标准。

失败样本库与提示词路由的价值

在这样的背景下，我们需要一种新的方法来应对LLM应用的测试挑战。本文提出的失败样本库驱动提示词路由持续迭代方法，正是为了解决这些问题而设计的。

这个方法的核心思想是：

1. 收集失败样本：将用户反馈或测试中发现的问题案例收集起来，构建一个失败样本库。
2. 提示词路由：根据输入特征，自动选择最合适的提示词模板来处理不同类型的请求。
3. 持续迭代：利用失败样本库不断优化提示词路由策略，形成一个闭环的改进过程。

这种方法的价值在于：

• 降低回归风险：通过保存失败案例，可以在每次系统变更后自动验证这些问题是否再次出现。
• 提高系统鲁棒性：针对不同类型的问题优化提示词，使系统能够更好地处理各种边缘情况。
• 加速迭代周期：将人工评估的重点放在新问题上，利用自动化处理已知问题，提高开发效率。
• 知识沉淀：失败样本库本身就是宝贵的知识库，记录了系统在实际使用中的问题和解决方案。

---

核心概念与理论基础

LLM应用回归测试的核心概念

在深入探讨解决方案之前，我们需要明确一些核心概念：

1. LLM应用的组件结构

一个典型的LLM应用通常包含以下组件：

在这个结构中，回归测试需要确保整个链路的稳定性，但最关键的变量在于提示词构建和LLM API调用环节。

2. LLM应用的"正确性"定义

与传统软件不同，LLM应用的"正确性"是多维度的：

• 事实正确性：输出内容是否符合事实，没有幻觉。
• 格式正确性：输出是否符合预期的格式要求。
• 相关性：输出是否与用户输入相关，没有偏离主题。
• 有用性：输出是否能够解决用户的问题或满足需求。
• 安全性：输出是否包含有害内容或敏感信息。

这些维度的权重取决于具体的应用场景，需要在测试中综合考虑。

3. 回归测试的覆盖范围

对于LLM应用，回归测试应该覆盖：

• 功能回归：确保核心功能在变更后仍然正常工作。
• 性能回归：确保响应时间、资源消耗等指标没有恶化。
• 质量回归：确保输出质量（如准确性、有用性）没有下降。
• 安全回归：确保系统仍然能够正确处理安全和内容政策问题。

失败样本库的设计原理

失败样本库是我们方法的核心组件之一，它不仅仅是一个问题集合，而是一个结构化的知识管理系统。

1. 样本的结构化表示

一个有效的失败样本应该包含以下信息：

class FailureSample:
    def __init__(self):
        self.sample_id = str  # 唯一标识符
        self.input_data = dict  # 原始输入数据
        self.expected_output = Optional[str]  # 预期输出（如果有）
        self.actual_output = str  # 实际失败输出
        self.failure_type = str  # 失败类型分类
        self.failure_severity = str  # 严重程度
        self.context = dict  # 上下文信息（模型版本、提示词版本等）
        self.timestamp = datetime  # 记录时间
        self.tags = List[str]  # 标签，用于分类和检索
        self.resolution_status = str  # 解决状态
        self.resolution_notes = Optional[str]  # 解决说明
        self.improved_prompt = Optional[str]  # 改进后的提示词

这种结构化表示使得样本可以被有效地管理、检索和分析。

2. 样本分类体系

为了更好地组织和利用失败样本，我们需要一个合理的分类体系：

失败类型	描述	示例
事实错误	输出包含虚假信息或幻觉	错误地回答某个公司成立于2000年，实际是1990年
格式错误	输出不符合预期格式	要求JSON格式但返回了纯文本
不相关	输出与输入问题无关	问天气但回答了食谱
不完整	输出内容不完整或被截断	代码示例只显示了一半
有害内容	输出包含不当或有害内容	生成暴力或歧视性内容
理解错误	模型误解了用户意图	将"翻译"理解为"解释"
性能问题	响应时间过长或超时	简单请求需要30秒以上

3. 样本库的价值流

失败样本库在整个系统中的价值流动如下：

提示词路由的基本原理

提示词路由是另一个核心概念，它解决的是"针对不同类型的输入，如何选择最合适的提示词"这个问题。

1. 为什么需要提示词路由

在实际应用中，我们经常会遇到以下情况：

• 不同类型的问题需要不同风格的提示词
• 某些提示词在处理特定问题时表现更好
• 模型对某些领域的知识有盲区，需要专门的提示词来弥补

提示词路由就是为了解决这些问题而设计的，它可以看作是LLM应用的"智能分发器"。

2. 提示词路由的核心要素

一个完整的提示词路由系统包含以下核心要素：

1. 输入分析器：分析用户输入，提取关键特征
2. 提示词模板库：存储针对不同场景优化的提示词模板
3. 路由规则库：定义特征与模板之间的映射规则
4. 路由决策引擎：根据输入特征和规则，选择最合适的模板
5. 模板实例化：将用户输入填充到选定的模板中，生成最终提示词

3. 路由策略的类型

常见的提示词路由策略包括：

1. 基于规则的路由：使用预定义的规则进行匹配

   • 关键词匹配
   • 正则表达式匹配
   • 语义相似度匹配

2. 基于模型的路由：使用ML模型进行分类

   • 文本分类模型
   • 相似度模型
   • 强化学习模型

3. 混合路由：结合规则和模型的优点

持续迭代的反馈循环机制

将失败样本库和提示词路由结合起来，我们可以构建一个持续迭代的反馈循环：

这个反馈循环的关键在于：

1. 问题驱动：从实际问题出发，确保优化方向与用户需求一致
2. 数据支撑：基于真实的失败样本进行决策，避免主观猜测
3. 持续改进：建立定期的分析和优化机制，使系统不断进化
4. 闭环验证：每次优化后都要进行回归测试，确保问题真正解决且没有引入新问题

---

环境准备

技术栈选择

为了实现我们的系统，我们选择以下技术栈：

组件	技术选择	版本要求	理由
编程语言	Python	3.9+	丰富的AI/ML生态系统，易于与LLM集成
Web框架	FastAPI	0.104+	高性能异步框架，自动生成API文档
数据库	SQLite/PostgreSQL	SQLite 3.35+ / PostgreSQL 13+	SQLite用于开发，PostgreSQL用于生产
向量数据库	ChromaDB	0.4.18+	轻量级，易于集成，支持语义搜索
LLM SDK	OpenAI Python SDK	1.3+	支持OpenAI和兼容OpenAI API的模型
数据处理	Pandas	2.0+	强大的数据处理和分析能力
前端 (可选)	Streamlit	1.28+	快速构建数据应用界面
测试框架	Pytest	7.4+	功能强大的Python测试框架

开发环境配置

1. 创建虚拟环境

# 使用venv创建虚拟环境
python -m venv llm-regression-env

# 激活虚拟环境
# Windows
llm-regression-env\Scripts\activate
# Linux/Mac
source llm-regression-env/bin/activate

2. 安装依赖库

创建一个requirements.txt文件：

fastapi==0.104.1
uvicorn==0.24.0.post1
sqlalchemy==2.0.23
chromadb==0.4.18
openai==1.3.5
pandas==2.1.3
scikit-learn==1.3.2
numpy==1.26.2
pydantic==2.5.2
python-dotenv==1.0.0
pytest==7.4.3
streamlit==1.28.2
python-multipart==0.0.6

安装依赖：

pip install -r requirements.txt

项目结构设计

我们将采用以下项目结构：

llm-regression-testing/
├── app/                      # 主应用目录
│   ├── __init__.py
│   ├── api/                  # API路由
│   │   ├── __init__.py
│   │   ├── samples.py        # 样本库API
│   │   ├── router.py         # 提示词路由API
│   │   └── testing.py        # 测试API
│   ├── core/                 # 核心功能模块
│   │   ├── __init__.py
│   │   ├── config.py         # 配置管理
│   │   ├── database.py       # 数据库连接
│   │   └── llm_client.py     # LLM客户端封装
│   ├── models/               # 数据模型
│   │   ├── __init__.py
│   │   ├── sample.py         # 样本模型
│   │   ├── prompt.py         # 提示词模板模型
│   │   └── route.py          # 路由规则模型
│   ├── services/             # 业务逻辑
│   │   ├── __init__.py
│   │   ├── sample_service.py # 样本库服务
│   │   ├── router_service.py # 路由服务
│   │   └── test_service.py   # 测试服务
│   └── utils/                # 工具函数
│       ├── __init__.py
│       ├── embedding.py      # 嵌入向量工具
│       ├── similarity.py     # 相似度计算
│       └── evaluator.py      # 输出评估工具
├── tests/                    # 测试代码
│   ├── __init__.py
│   ├── test_sample_service.py
│   └── test_router_service.py
├── data/                     # 数据目录
│   ├── samples/              # 样本数据
│   └── prompts/              # 提示词模板
├── notebooks/                # Jupyter notebooks
├── scripts/                  # 脚本工具
├── .env.example              # 环境变量示例
├── requirements.txt          # 依赖列表
├── main.py                   # 应用入口
└── README.md                 # 项目说明

配置环境变量

创建.env文件：

# 应用配置
APP_NAME=LLM Regression Testing Tool
APP_VERSION=1.0.0
DEBUG=True

# 数据库配置
DATABASE_URL=sqlite:///./data/llm_testing.db

# ChromaDB配置
CHROMA_DB_PATH=./data/chroma

# LLM配置
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
DEFAULT_MODEL=gpt-3.5-turbo
EMBEDDING_MODEL=text-embedding-ada-002

# 路由配置
DEFAULT_ROUTING_STRATEGY=hybrid
SIMILARITY_THRESHOLD=0.7

---

分步实现：失败样本库

样本数据模型设计

首先，我们需要定义样本数据的模型。我们将使用SQLAlchemy作为ORM框架。

创建app/models/sample.py：

from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional
from sqlalchemy import Column, String, DateTime, Text, Integer, JSON, Enum as SQLEnum
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class FailureType(Enum):
    FACTUAL_ERROR = "factual_error"
    FORMAT_ERROR = "format_error"
    IRRELEVANT = "irrelevant"
    INCOMPLETE = "incomplete"
    HARMFUL_CONTENT = "harmful_content"
    MISUNDERSTANDING = "misunderstanding"
    PERFORMANCE_ISSUE = "performance_issue"
    OTHER = "other"


class Severity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class ResolutionStatus(Enum):
    OPEN = "open"
    ANALYZING = "analyzing"
    RESOLVED = "resolved"
    CLOSED = "closed"
    WONT_FIX = "wont_fix"


class FailureSample(Base):
    __tablename__ = "failure_samples"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    sample_id = Column(String, unique=True, index=True, nullable=False)
    
    # 输入输出数据
    input_data = Column(JSON, nullable=False)
    expected_output = Column(Text, nullable=True)
    actual_output = Column(Text, nullable=False)
    
    # 元数据
    failure_type = Column(SQLEnum(FailureType), nullable=False)
    severity = Column(SQLEnum(Severity), nullable=False, default=Severity.MEDIUM)
    context = Column(JSON, nullable=True)  # 模型版本、提示词版本等
    
    # 时间戳
    timestamp = Column(DateTime, default=datetime.utcnow, nullable=False)
    resolved_at = Column(DateTime, nullable=True)
    
    # 分类和状态
    tags = Column(JSON, nullable=True)  # List[str]
    resolution_status = Column(SQLEnum(ResolutionStatus), nullable=False, default=ResolutionStatus.OPEN)
    resolution_notes = Column(Text, nullable=True)
    improved_prompt = Column(Text, nullable=True)
    
    # 关联信息
    related_sample_ids = Column(JSON, nullable=True)  # List[str]
    
    def __repr__(self):
        return f"<FailureSample {self.sample_id} {self.failure_type}>"

数据采集与标注流程

接下来，我们需要实现样本采集和标注的功能。创建app/services/sample_service.py：

import uuid
from datetime import datetime
from typing import List, Optional, Dict, Any
from sqlalchemy.orm import Session
from sqlalchemy import or_, and_

from app.models.sample import (
    FailureSample, FailureType, Severity, ResolutionStatus, Base
)
from app.core.database import SessionLocal, engine


class SampleService:
    def __init__(self, db: Session):
        self.db = db
    
    def create_sample(
        self,
        input_data: Dict[str, Any],
        actual_output: str,
        failure_type: FailureType,
        expected_output: Optional[str] = None,
        severity: Severity = Severity.MEDIUM,
        context: Optional[Dict[str, Any]] = None,
        tags: Optional[List[str]] = None
    ) -> FailureSample:
        """创建新的失败样本"""
        sample_id = f"sample-{uuid.uuid4().hex[:8]}"
        
        sample = FailureSample(
            sample_id=sample_id,
            input_data=input_data,
            actual_output=actual_output,
            failure_type=failure_type,
            expected_output=expected_output,
            severity=severity,
            context=context or {},
            tags=tags or [],
            resolution_status=ResolutionStatus.OPEN
        )
        
        self.db.add(sample)
        self.db.commit()
        self.db.refresh(sample)
        
        return sample
    
    def get_sample(self, sample_id: str) -> Optional[FailureSample]:
        """根据ID获取样本"""
        return self.db.query(FailureSample).filter(
            FailureSample.sample_id == sample_id
        ).first()
    
    def list_samples(
        self,
        skip: int = 0,
        limit: int = 100,
        failure_type: Optional[FailureType] = None,
        severity: Optional[Severity] = None,
        status: Optional[ResolutionStatus] = None,
        tag: Optional[str] = None,
        search_query: Optional[str] = None
    ) -> List[FailureSample]:
        """列出样本，支持过滤和搜索"""
        query = self.db.query(FailureSample)
        
        # 应用过滤条件
        if failure_type:
            query = query.filter(FailureSample.failure_type == failure_type)
        if severity:
            query = query.filter(FailureSample.severity == severity)
        if status:
            query = query.filter(FailureSample.resolution_status == status)
        if tag:
            query = query.filter(FailureSample.tags.contains([tag]))
        
        # 搜索功能
        if search_query:
            query = query.filter(
                or_(
                    FailureSample.input_data.cast(String).contains(search_query),
                    FailureSample.actual_output.contains(search_query),
                    FailureSample.sample_id.contains(search_query)
                )
            )
        
        # 排序和分页
        query = query.order_by(FailureSample.timestamp.desc())
        query = query.offset(skip).limit(limit)
        
        return query.all()
    
    def update_sample(
        self,
        sample_id: str,
        **kwargs
    ) -> Optional[FailureSample]:
        """更新样本信息"""
        sample = self.get_sample(sample_id)
        if not sample:
            return None
        
        # 更新字段
        for key, value in kwargs.items():
            if hasattr(sample, key) and value is not None:
                setattr(sample, key, value)
        
        # 如果状态变为已解决，记录解决时间
        if kwargs.get("resolution_status") == ResolutionStatus.RESOLVED:
            sample.resolved_at = datetime.utcnow()
        
        self.db.commit()
        self.db.refresh(sample)
        return sample
    
    def add_tag(self, sample_id: str, tag: str) -> Optional[FailureSample]:
        """为样本添加标签"""
        sample = self.get_sample(sample_id)
        if not sample:
            return None
        
        if tag not in sample.tags:
            sample.tags.append(tag)
            self.db.commit()
            self.db.refresh(sample)
        
        return sample
    
    def remove_tag(self, sample_id: str, tag: str) -> Optional[FailureSample]:
        """移除样本标签"""
        sample = self.get_sample(sample_id)
        if not sample:
            return None
        
        if tag in sample.tags:
            sample.tags.remove(tag)
            self.db.commit()
            self.db.refresh(sample)
        
        return sample
    
    def get_statistics(self) -> Dict[str, Any]:
        """获取样本库统计信息"""
        total = self.db.query(FailureSample).count()
        
        # 按失败类型统计
        type_stats = {}
        for ft in FailureType:
            count = self.db.query(FailureSample).filter(
                FailureSample.failure_type == ft
            ).count()
            type_stats[ft.value] = count
        
        # 按状态统计
        status_stats = {}
        for rs in ResolutionStatus:
            count = self.db.query(FailureSample).filter(
                FailureSample.resolution_status == rs
            ).count()
            status_stats[rs.value] = count
        
        # 按严重程度统计
        severity_stats = {}
        for sev in Severity:
            count = self.db.query(FailureSample).filter(
                FailureSample.severity == sev
            ).count()
            severity_stats[sev.value] = count
        
        return {
            "total_samples": total,
            "by_failure_type": type_stats,
            "by_resolution_status": status_stats,
            "by_severity": severity_stats
        }


# 数据库初始化函数
def init_db():
    Base.metadata.create_all(bind=engine)

样本库存储方案

我们将使用SQLite存储结构化数据，使用ChromaDB存储向量数据以便进行语义搜索。

创建app/core/database.py：

import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import chromadb

# 加载环境变量
load_dotenv()

# SQLAlchemy配置
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./data/llm_testing.db")

# 创建数据库引擎
engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


# ChromaDB配置
CHROMA_DB_PATH = os.getenv("CHROMA_DB_PATH", "./data/chroma")

# 确保数据目录存在
os.makedirs(os.path.dirname(CHROMA_DB_PATH), exist_ok=True)

# 初始化ChromaDB客户端
chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH)


# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

样本库管理功能实现

接下来，我们创建API端点来提供样本库的管理功能。创建app/api/samples.py：

from typing import List, Optional, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.models.sample import FailureType, Severity, ResolutionStatus
from app.services.sample_service import SampleService

router = APIRouter(prefix="/api/samples", tags=["samples"])


def get_sample_service(db: Session = Depends(get_db)) -> SampleService:
    return SampleService(db)


@router.post("/")
def create_sample(
    input_data: Dict[str, Any],
    actual_output: str,
    failure_type: FailureType,
    expected_output: Optional[str] = None,
    severity: Severity = Severity.MEDIUM,
    context: Optional[Dict[str, Any]] = None,
    tags: Optional[List[str]] = None,
    sample_service: SampleService = Depends(get_sample_service)
):
    """创建新的失败样本"""
    try:
        sample = sample_service.create_sample(
            input_data=input_data,
            actual_output=actual_output,
            failure_type=failure_type,
            expected_output=expected_output,
            severity=severity,
            context=context,
            tags=tags
        )
        return {
            "success": True,
            "data": {
                "sample_id": sample.sample_id,
                "message": "Sample created successfully"
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.get("/")
def list_samples(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    failure_type: Optional[FailureType] = None,
    severity: Optional[Severity] = None,
    status: Optional[ResolutionStatus] = None,
    tag: Optional[str] = None,
    search_query: Optional[str] = None,
    sample_service: SampleService = Depends(get_sample_service)
):
    """列出样本，支持过滤和搜索"""
    samples = sample_service.list_samples(
        skip=skip,
        limit=limit,
        failure_type=failure_type,
        severity=severity,
        status=status,
        tag=tag,
        search_query=search_query
    )
    
    # 转换为字典格式
    sample_dicts = []
    for sample in samples:
        sample_dict = {
            "sample_id": sample.sample_id,
            "input_data": sample.input_data,
            "actual_output": sample.actual_output,
            "failure_type": sample.failure_type.value,
            "severity": sample.severity.value,
            "resolution_status": sample.resolution_status.value,
            "tags": sample.tags,
            "timestamp": sample.timestamp.isoformat() if sample.timestamp else None
        }
        if sample.expected_output:
            sample_dict["expected_output"] = sample.expected_output
        if sample.context:
            sample_dict["context"] = sample.context
        sample_dicts.append(sample_dict)
    
    return {
        "success": True,
        "data": sample_dicts,
        "pagination": {
            "skip": skip,
            "limit": limit,
            "total": len(sample_dicts)
        }
    }


@router.get("/{sample_id}")
def get_sample(
    sample_id: str,
    sample_service: SampleService = Depends(get_sample_service)
):
    """获取单个样本的详细信息"""
    sample = sample_service.get_sample(sample_id)
    if not sample:
        raise HTTPException(status_code=404, detail="Sample not found")
    
    return {
        "success": True,
        "data": {
            "sample_id": sample.sample_id,
            "input_data": sample.input_data,
            "actual_output": sample.actual_output,
            "expected_output": sample.expected_output,
            "failure_type": sample.failure_type.value,
            "severity": sample.severity.value,
            "context": sample.context,
            "timestamp": sample.timestamp.isoformat() if sample.timestamp else None,
            "tags": sample.tags,
            "resolution_status": sample.resolution_status.value,
            "resolution_notes": sample.resolution_notes,
            "improved_prompt": sample.improved_prompt,
            "resolved_at": sample.resolved_at.isoformat() if sample.resolved_at else None,
            "related_sample_ids": sample.related_sample_ids
        }
    }


@router.put("/{sample_id}")
def update_sample(
    sample_id: str,
    resolution_status: Optional[ResolutionStatus] = None,
    resolution_notes: Optional[str] = None,
    improved_prompt: Optional[str] = None,
    severity: Optional[Severity] = None,
    failure_type: Optional[FailureType] = None,
    sample_service: SampleService = Depends(get_sample_service)
):
    """更新样本信息"""
    update_data = {}
    if resolution_status is not None:
        update_data["resolution_status"] = resolution_status
    if resolution_notes is not None:
        update_data["resolution_notes"] = resolution_notes
    if improved_prompt is not None:
        update_data["improved_prompt"] = improved_prompt
    if severity is not None:
        update_data["severity"] = severity
    if failure_type is not None:
        update_data["failure_type"] = failure_type
    
    if not update_data:
        raise HTTPException(status_code=400, detail="No update data provided")
    
    sample = sample_service.update_sample(sample_id, **update_data)
    if not sample:
        raise HTTPException(status_code=404, detail="Sample not found")
    
    return {
        "success": True,
        "data": {
            "sample_id": sample.sample_id,
            "message": "Sample updated successfully"
        }
    }


@router.post("/{sample_id}/tags")
def add_tag(
    sample_id: str,
    tag: str,
    sample_service: SampleService = Depends(get_sample_service)
):
    """为样本添加标签"""
    sample = sample_service.add_tag(sample_id, tag)
    if not sample:
        raise HTTPException(status_code=404, detail="Sample not found")
    
    return {
        "success": True,
        "data": {
            "sample_id": sample.sample_id,
            "tags": sample.tags,
            "message": "Tag added successfully"
        }
    }


@router.delete("/{sample_id}/tags/{tag}")
def remove_tag(
    sample_id: str,
    tag: str,
    sample_service: SampleService = Depends(get_sample_service)
):
    """移除样本标签"""
    sample = sample_service.remove_tag(sample_id, tag)
    if not sample:
        raise HTTPException(status_code=404, detail="Sample not found")
    
    return {
        "success": True,
        "data": {
            "sample_id": sample.sample_id,
            "tags": sample.tags,
            "message": "Tag removed successfully"
        }
    }


@router.get("/statistics/overview")
def get_statistics(
    sample_service: SampleService = Depends(get_sample_service)
):
    """获取样本库统计信息"""
    stats = sample_service.get_statistics()
    return {
        "success": True,
        "data": stats
    }

---

分步实现：提示词路由工具

路由策略设计

提示词路由的核心是根据输入特征选择合适的提示词模板。我们将设计一个灵活的路由系统，支持多种路由策略。

首先，创建提示词模板和路由规则的数据模型。创建app/models/prompt.py：

from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional, Any
from sqlalchemy import Column, String, DateTime, Text, Integer, JSON, Boolean, Enum as SQLEnum, ForeignKey
from sqlalchemy.orm import relationship
from app.models.sample import Base


class TemplateType(Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"
    FUNCTION = "function"


class RoutingStrategy(Enum):
    KEYWORD = "keyword"
    SEMANTIC = "semantic"
    RULE_BASED = "rule_based"
    HYBRID = "hybrid"
    ML_BASED = "ml_based"


class PromptTemplate(Base):
    __tablename__ = "prompt_templates"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    template_id = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False)
    description = Column(Text, nullable=True)
    
    # 模板内容
    template_content = Column(Text, nullable=False)
    template_type = Column(SQLEnum(TemplateType), nullable=False, default=TemplateType.SYSTEM)
    
    # 模板变量
    variables = Column(JSON, nullable=True)  # List[str] - 变量名列表
    
    # 分类和标签
    category = Column(String, nullable=True)
    tags = Column(JSON, nullable=True)  # List[str]
    
    # 元数据
    version = Column(Integer, default=1, nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
    created_by = Column(String, nullable=True)
    
    # 使用统计
    usage_count = Column(Integer, default=0, nullable=False)
    success_rate = Column(Integer, default=0, nullable=False)  # 0-100
    
    # 关联路由规则
    routing_rules = relationship("RoutingRule", back_populates="template")
    
    def __repr__(self):
        return f"<PromptTemplate {self.template_id} v{self.version}>"


class RoutingRule(Base):
    __tablename__ = "routing_rules"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    rule_id = Column(String, unique=True, index=True, nullable=False)
    name = Column(String, nullable=False)
    description = Column(Text, nullable=True)
    
    # 关联模板
    template_id = Column(String, ForeignKey("prompt_templates.template_id"), nullable=False)
    template = relationship("PromptTemplate", back_populates="routing_rules")
    
    # 路由条件
    priority = Column(Integer, default=0, nullable=False)  # 优先级，数字越大优先级越高
    conditions = Column(JSON, nullable=False)  # 条件定义
    
    # 规则状态
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
    
    def __repr__(self):
        return f"<RoutingRule {self.rule_id}>"

提示词模板管理

接下来，我们实现提示词模板的管理功能。创建app/services/router_service.py：

import uuid
import re
from datetime import datetime
from typing import List, Optional, Dict, Any, Tuple
from sqlalchemy.orm import Session
from sqlalchemy import or_, and_, desc

from app.models.prompt import (
    PromptTemplate, RoutingRule, TemplateType, RoutingStrategy, Base
)
from app.core.database import SessionLocal, engine
from app.utils.embedding import get_embedding, compute_similarity


class PromptTemplateService:
    def __init__(self, db: Session):
        self.db = db
    
    def create_template(
        self,
        name: str,
        template_content: str,
        template_type: TemplateType = TemplateType.SYSTEM,
        description: Optional[str] = None,
        variables: Optional[List[str]] = None,
        category: Optional[str] = None,
        tags: Optional[List[str]] = None,
        created_by: Optional[str] = None
    ) -> PromptTemplate:
        """创建新的提示词模板"""
        template_id = f"template-{uuid.uuid4().hex[:8]}"
        
        # 自动提取变量（如果没有提供）
        if variables is None:
            variables = self._extract_variables(template_content)
        
        template = PromptTemplate(
            template_id=template_id,
            name=name,
            description=description,
            template_content=template_content,
            template_type=template_type,
            variables=variables,
            category=category,
            tags=tags or [],
            created_by=created_by
        )
        
        self.db.add(template)
        self.db.commit()
        self.db.refresh(template)
        
        return template
    
    def _extract_variables(self, template_content: str) -> List[str]:
        """从模板内容中提取变量名（格式：{variable_name}）"""
        pattern = r'\{([a-zA-Z_][a-zA-Z0-9_]*)\}'
        variables = re.findall(pattern, template_content)
        # 去重并保持顺序
        seen = set()
        unique_vars = []
        for var in variables:
            if var not in seen:
                seen.add(var)
                unique_vars.append(var)
        return unique_vars
    
    def get_template(self, template_id: str) -> Optional[PromptTemplate]:
        """根据ID获取模板"""
        return self.db.query(PromptTemplate).filter(
            and_(
                PromptTemplate.template_id == template_id,
                PromptTemplate.is_active == True
            )
        ).first()
    
    def list_templates(
        self,
        skip: int = 0,
        limit: int = 100,
        template_type: Optional[TemplateType] = None,
        category: Optional[str] = None,
        tag: Optional[str] = None,
        search_query: Optional[str] = None
    ) -> List[PromptTemplate]:
        """列出模板，支持过滤和搜索"""
        query = self.db.query(PromptTemplate).filter(
            PromptTemplate.is_active == True
        )
        
        # 应用过滤条件
        if template_type:
            query = query.filter(PromptTemplate.template_type == template_type)
        if category:
            query = query.filter(PromptTemplate.category == category)
        if tag:
            query = query.filter(PromptTemplate.tags.contains([tag]))
        
        # 搜索功能
        if search_query:
            query = query.filter(
                or_(
                    PromptTemplate.name.contains(search_query),
                    Prompt