解析 ‘Regression Testing’：在更新 LangChain 版本后，如何确保原有的业务逻辑没有发生语义偏移？

各位同仁，各位技术探索者，大家好！

今天，我们齐聚一堂，共同探讨一个在快速迭代的AI时代中日益凸显的议题：如何在更新 LangChain 这样的核心库版本后，确保我们辛辛苦苦构建的业务逻辑没有发生“语义偏移”。这不仅仅是代码层面的兼容性问题，更深层次地，它触及到我们基于大型语言模型（LLM）的应用赖以生存的“智能”核心。

想象一下，你精心设计了一个复杂的问答系统，一个智能客服代理，或者一个内容生成流水线。它们在旧版本的 LangChain 上运行得天衣无缝。然而，当新的 LangChain 版本发布，带来了性能优化、新功能或者对现有模块的重构时，你满怀期待地升级了。但随之而来的，可能是潜伏的风险：原先清晰准确的回答变得模糊，代理的决策逻辑开始偏离，甚至某些特定输入下的行为完全出乎意料。这就是我们所说的“语义偏移”，它像一个无形的幽灵，可能悄无声息地侵蚀你的应用质量。

在传统的软件开发中，回归测试 (Regression Testing) 是确保功能稳定的基石。但在LLM驱动的应用中，由于其固有的非确定性、对底层模型行为的敏感性以及输出的开放性，回归测试变得尤为复杂和关键。今天，我将以一名编程专家的视角，为大家深入剖析如何在 LangChain 版本升级后，构建一套严谨、高效的回归测试策略，确保我们的应用能够持续提供可靠、稳定的服务。

我们将从理解语义偏移的本质开始，剖析 LangChain 升级可能带来的具体影响，然后深入探讨测试原则、环境搭建，并最终落实在具体的测试策略、代码实践以及推荐工具上。

---

第一讲：理解LLM应用中的“语义偏移”

在深入探讨测试方法之前，我们必须首先明确“语义偏移”在LLM应用中的具体含义。它与传统软件的“功能失效”有所不同。

传统软件的功能失效通常是二元对立的：一个按钮要么能点击，要么不能；一个函数要么返回正确结果，要么报错或返回错误结果。这种失效是确定性的、可重现的。

LLM应用中的语义偏移则更加微妙和难以捉摸。它指的是：

1. 输出内容的变化 (Content Shift):

   • 准确性下降： 原本正确的答案变得不准确或包含错误信息。
   • 完整性缺失： 答案遗漏了关键信息。
   • 风格/语气变化： 原本友好专业的回复变得生硬、随意，甚至带有偏见。
   • 冗余性增加或减少： 答案变得过于冗长或过于简短，不符合预期。
   • 格式破坏： 预期中的Markdown列表变成了纯文本，或者JSON格式被破坏。

2. 决策逻辑的变化 (Logic Shift):

   • 代理行为异常： Agent 在面对特定任务时，选择了错误的工具，或者推理路径变得低效、不合理。
   • 链式反应中断： 原本设计好的多步骤链（Chain）在某个环节的输出不再符合下一个环节的输入预期，导致整个流程中断或产生错误结果。
   • 检索相关性下降： RAG（Retrieval Augmented Generation）系统中，检索到的文档不再高度相关，导致最终生成的内容质量下降。

3. 性能指标的变化 (Performance Shift):

   • 延迟增加： 相同任务的响应时间显著延长。
   • 成本升高： 由于不必要的LLM调用或更长的输入/输出，导致API调用成本上升。
   • 稳定性下降： 以前罕见的错误或超时变得频繁。

为什么LLM应用更容易发生语义偏移？

• LLM的非确定性： 即使是同一个模型，在相同的输入下，也可能产生略有不同的输出（尤其是在温度参数较高时）。LangChain的更新可能会改变与LLM交互的方式，例如默认参数、重试机制等，从而间接影响LLM的输出。
• Prompt的敏感性： LLM对Prompt的微小变化极其敏感。LangChain内部对Prompt模板的渲染方式、消息历史的构建、工具描述的生成等任何细微调整，都可能导致LLM的理解和行为发生巨大变化。
• 依赖的链式效应： LangChain应用的复杂性在于其组件（模型、Prompt、解析器、工具、检索器、Agent）之间的链式依赖。一个底层组件的微小变化，都可能在整个链条上传导并放大，最终导致意想不到的语义偏移。
• 底层LLM模型的变化： 即使LangChain版本不变，我们使用的OpenAI、Anthropic等LLM服务提供商也可能更新他们的模型版本（例如 gpt-3.5-turbo-0613 到 gpt-3.5-turbo-1106）。这些底层模型的更新本身就可能带来行为上的变化，而LangChain的更新可能会更好地适配这些新模型，但也可能因此暴露出旧逻辑与新模型的不兼容。

理解这些，是我们构建有效回归测试策略的前提。

---

第二讲：LangChain升级的潜在影响面分析

LangChain作为一个快速发展的框架，其版本迭代非常频繁。每一次更新都可能带来显著的改变。作为开发者，我们需要对这些潜在影响面有清晰的认识，才能有针对性地进行测试。

LangChain更新的常见类型及影响：

1. API/接口变更 (Breaking Changes):

   • 影响： 最直接的影响，旧代码无法编译或运行。这通常会在升级后立即发现。例如，某个类的构造函数参数改变，某个方法的名称或签名改变。
   • 示例： BasePromptTemplate 的 format 方法签名变化，或者 LLMChain 的实例化方式改变。

2. 内部实现逻辑重构 (Internal Logic Refactoring):

   • 影响： 代码可能继续运行，但行为可能发生变化。例如，Agent的决策逻辑、RAG的检索策略、输出解析器的处理方式等。这是语义偏移最常见的温床。
   • 示例： AgentExecutor 内部选择工具的算法优化；RecursiveCharacterTextSplitter 的默认参数调整；ConversationalRetrievalChain 处理历史对话的方式改变。

3. 默认参数变更 (Default Parameter Changes):

   • 影响： 即使不修改代码，组件的行为也可能因默认参数的改变而不同。这可能导致性能、成本或输出质量的隐性变化。
   • 示例： ChatOpenAI 的默认 temperature 改变；VectorStoreRetriever 的默认 k (检索数量) 改变。

4. 新模块/功能引入 (New Modules/Features):

   • 影响： 通常是正面的，但如果新的模块取代了旧的推荐用法，旧代码虽然能跑，但可能不再是最佳实践，甚至与新模块产生冲突。
   • 示例： LCEL (LangChain Expression Language) 的引入，使得构建链的方式更加灵活高效，但旧的 LLMChain, StuffDocumentsChain 等虽然仍可用，其组合方式可能不再推荐。

5. 依赖库版本升级 (Dependency Updates):

   • 影响： LangChain本身依赖于大量的第三方库（如 openai, tiktoken, pydantic）。这些依赖库的升级也可能带来其自身的兼容性问题或行为变化，并通过LangChain传递到你的应用。
   • 示例： pydantic 从 V1 升级到 V2，其模型验证和序列化行为可能发生显著变化，影响LangChain内部使用Pydantic进行数据模型定义和解析的部分。

核心组件的潜在影响点：

为了更系统地评估风险，我们可以将 LangChain 应用解构为几个核心组件，并考虑每个组件在升级后可能发生语义偏移的地方。

组件类型	潜在影响点	示例
LLMs/ChatModels	默认参数 (temperature, top_p等)，API调用方式，消息格式兼容性，错误处理。	ChatOpenAI 默认 model_name 改变，temperature 从 0.7 变为 0.5。
Prompt Templates	变量渲染逻辑，消息结构构建，特殊字符处理，模板语言（Jinja2, F-string）更新。	ChatPromptTemplate.from_messages 构建 SystemMessage 或 HumanMessage 的内部逻辑变化。
Chains/Runnables	内部调用顺序，中间结果传递，错误传播，LCEL表达式的解析和执行。	SequentialChain 中间变量传递问题；RunnableWithMessageHistory 的会话管理机制调整。
Tools	工具描述生成方式，输入参数解析，输出格式约定，tool_code 定义。	Tool 类构造函数参数变化；StructuredTool 的 args_schema 验证逻辑更新。
Agents	决策逻辑，工具选择算法，思考路径（reasoning trace）生成，输出解析（AgentOutputParser）。	AgentExecutor 的 handle_parsing_errors 行为改变；ReAct 代理的思考步骤生成优化。
Retrievers	文档加载器 (Document Loaders)，文本分割器 (Text Splitters) 的默认参数，嵌入模型 (Embeddings)，向量存储 (VectorStores) 的查询逻辑，相关性排序。	RecursiveCharacterTextSplitter 默认 chunk_size 变化；FAISS 向量存储查询默认 k 值。
Output Parsers	解析复杂结构（如JSON, CSV）的鲁棒性，错误处理，特定格式（如列表、对象）的识别。	PydanticOutputParser 对 pydantic V2 的适配；CommaSeparatedListOutputParser 对空格的处理。
Callbacks	回调事件触发时机，回调函数参数签名，异步回调处理。	BaseCallbackHandler 中的 on_llm_start 或 on_tool_end 方法参数变化。

有了这个影响面清单，我们就能更系统地设计测试用例，覆盖到最可能发生问题的区域。

---

第三讲：LLM应用回归测试的核心原则与方法论

LLM应用的回归测试，在继承传统软件测试严谨性的同时，必须引入针对非确定性和语义的特殊考量。

核心原则：

1. 分层测试： 像传统软件一样，采用单元测试、集成测试和端到端测试相结合的方式。

   • 单元测试 (Unit Tests): 关注单个LangChain组件（如自定义Prompt模板、自定义工具、自定义解析器）的独立功能。
   • 集成测试 (Integration Tests): 关注LangChain组件之间的协作（如Chain、Agent与工具的组合）。这是发现语义偏移的关键层。
   • 端到端测试 (End-to-End Tests): 模拟真实用户场景，验证整个应用流程。

2. 基线（Golden Set）建立： 在升级前，为关键的输入-输出对建立“黄金标准”或“基线”。这些基线代表了“已知良好”的行为。升级后，将新版本的输出与这些基线进行比较。

   • 基线的形式： 可以是精确的字符串、JSON结构、布尔值，也可以是更宽松的语义评估（如相关性评分、关键词存在）。

3. 多维度评估： 不仅仅检查输出是否“完全相同”，还要评估其“语义相似性”、“准确性”、“完整性”、“风格”以及“性能指标”（延迟、成本）。

4. 自动化与人工结合： 大部分常规检查应自动化，但对于涉及主观判断、复杂推理的场景，人工评审仍然不可或缺。LLM-as-a-Judge (使用另一个LLM进行评估) 是一种半自动化方案。

5. 版本控制一切： 不仅仅是代码，还包括：

   • Prompt模板： 确保Prompt文本被版本控制。
   • 测试数据： 明确每个测试用例的输入和期望输出。
   • 模型版本： 记录测试时使用的LLM模型名称和版本。

6. 拥抱非确定性，但加以约束：

   • 阈值与范围： 对于数值型或语义相似性指标，接受在一个可容忍的范围内波动，而不是严格相等。
   • 多次运行与统计分析： 对于不稳定的LLM输出，可以多次运行测试，并分析结果的统计分布（如平均值、标准差），而非单次结果。

测试环境搭建：

一个健壮的测试环境是高效回归测试的基础。

1. 依赖管理：

   • 使用 pip freeze > requirements.txt 冻结当前所有依赖版本，并在升级前保存一份。
   • 使用虚拟环境（venv 或 conda）隔离不同项目的依赖，或者在升级时创建新的虚拟环境。

   # 在升级前
   python -m venv old_env
   source old_env/bin/activate
   pip install -r requirements.txt # 安装旧版本依赖
   pip freeze > requirements_old.txt
   deactivate
   
   # 准备升级
   python -m venv new_env
   source new_env/bin/activate
   # 编辑 requirements.txt，将 langchain 版本更新
   # 例如：langchain==0.0.350 -> langchain==0.1.0
   pip install -r requirements.txt

2. 版本控制：

   • 在进行大规模升级前，务必创建专门的分支（例如 feat/upgrade-langchain-v0.1）。
   • 将所有测试代码、测试数据、基线数据都纳入版本控制。

3. 配置管理：

   • 将LLM API密钥、模型名称、数据库连接字符串等敏感信息和可变配置项通过环境变量、.env 文件或专门的配置管理工具（如 python-dotenv, dynaconf）进行管理，避免硬编码。
   • 在测试环境中，可以使用模拟的API密钥或指向本地/测试服务的URL。

   # config.py
   import os
   from dotenv import load_dotenv
   
   load_dotenv() # 加载 .env 文件中的环境变量
   
   OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
   OPENAI_MODEL_NAME = os.getenv("OPENAI_MODEL_NAME", "gpt-3.5-turbo")

4. 测试数据管理：

   • 输入数据集： 准备一份具有代表性、覆盖各种业务场景的输入数据集。包括：
     • Happy Path (正常路径): 预期能顺利处理的常见输入。
     • Edge Cases (边缘情况): 边界值、特殊字符、空输入、非常规但合法的输入。
     • Negative Cases (异常情况): 错误格式、恶意输入、超出能力范围的输入，验证错误处理机制。
   • 基线（Golden Dataset）： 对于每个输入，存储旧版本应用生成的“正确”或“可接受”的输出。这可以是简单的文本文件、JSON文件，甚至是数据库记录。

   # tests/golden_data/qa_chain_v0_0_350.json
   [
       {
           "input": "LangChain是什么？",
           "expected_output": "LangChain是一个开源框架，用于开发由大型语言模型驱动的应用程序。"
       },
       {
           "input": "请总结一下人工智能的最新进展。",
           "expected_output": "人工智能在深度学习、自然语言处理和计算机视觉等领域取得了显著进展，特别是在大模型和生成式AI方面。"
       }
   ]

---

第四讲：LangChain回归测试的具体策略与代码实践

现在，让我们深入到具体的测试策略和代码实践中，涵盖单元测试、集成测试和端到端测试。

我们将使用 pytest 作为测试框架，因为它简洁灵活，并广泛应用于Python社区。

4.1 单元测试：确保组件基础功能稳定

单元测试主要关注自定义的 LangChain 组件，例如自定义工具、自定义解析器、Prompt 模板的渲染等。

4.1.1 Prompt Template 渲染测试

确保 Prompt 模板在不同变量下能正确渲染，且结构符合预期。

# app/prompts.py
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate

def get_qa_prompt_template():
    return ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template("你是一个友好的AI助手。请根据以下上下文回答问题：nn{context}"),
        HumanMessagePromptTemplate.from_template("问题：{question}")
    ])

# tests/test_prompts.py
import pytest
from app.prompts import get_qa_prompt_template
from langchain_core.messages import HumanMessage, SystemMessage

def test_qa_prompt_template_rendering():
    prompt_template = get_qa_prompt_template()

    context = "LangChain是一个开源框架。"
    question = "LangChain是什么？"

    # 渲染Prompt
    messages = prompt_template.format_messages(context=context, question=question)

    assert len(messages) == 2
    assert isinstance(messages[0], SystemMessage)
    assert isinstance(messages[1], HumanMessage)

    assert messages[0].content == "你是一个友好的AI助手。请根据以下上下文回答问题：nnLangChain是一个开源框架。"
    assert messages[1].content == "问题：LangChain是什么？"

    # 验证缺少变量时是否报错（或有预期行为）
    with pytest.raises(KeyError): # 或者其他 LangChain 内部定义的错误
        prompt_template.format_messages(question=question)

4.1.2 自定义工具 (Custom Tool) 测试

验证自定义工具的输入、输出以及内部逻辑是否符合预期。

# app/tools.py
from langchain.tools import BaseTool
import requests

class WeatherTool(BaseTool):
    name = "天气查询工具"
    description = "用于查询指定城市当前天气信息的工具。输入应为城市名称，例如'北京'。"

    def _run(self, city: str) -> str:
        try:
            # 模拟API调用，实际应用中会调用真实天气API
            if city == "北京":
                return "北京今天晴，气温25°C，微风。"
            elif city == "上海":
                return "上海今天多云，气温28°C，阵雨。"
            else:
                return f"抱歉，暂无{city}的天气信息。"
        except Exception as e:
            return f"查询天气时发生错误: {e}"

    async def _arun(self, city: str) -> str:
        # 异步版本
        return self._run(city)

# tests/test_tools.py
import pytest
from app.tools import WeatherTool

def test_weather_tool_valid_city():
    tool = WeatherTool()
    result = tool.run("北京")
    assert "晴" in result
    assert "25°C" in result

def test_weather_tool_invalid_city():
    tool = WeatherTool()
    result = tool.run("火星")
    assert "暂无火星的天气信息" in result

def test_weather_tool_error_handling(mocker):
    # 使用 mocker 模拟内部请求失败
    mocker.patch('app.tools.requests.get', side_effect=requests.exceptions.RequestException("网络错误"))
    tool = WeatherTool()
    result = tool.run("深圳")
    assert "查询天气时发生错误" in result

4.1.3 自定义输出解析器 (Output Parser) 测试

确保解析器能够正确地从LLM的原始输出中提取结构化信息。

# app/parsers.py
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

class AnswerAndSources(BaseModel):
    answer: str = Field(description="The answer to the user's question.")
    sources: list[str] = Field(description="A list of source documents used to answer the question.")

class CustomJsonParser(JsonOutputParser):
    def get_format_instructions(self) -> str:
        return "请以JSON格式输出，包含'answer'和'sources'字段。"

# tests/test_parsers.py
import pytest
from app.parsers import CustomJsonParser, AnswerAndSources

def test_custom_json_parser_valid_json():
    parser = CustomJsonParser(pydantic_object=AnswerAndSources)
    llm_output = '```jsonn{"answer": "LangChain是一个框架。", "sources": ["文档A", "文档B"]}n```'

    parsed_output = parser.parse(llm_output)

    assert isinstance(parsed_output, AnswerAndSources)
    assert parsed_output.answer == "LangChain是一个框架。"
    assert parsed_output.sources == ["文档A", "文档B"]

def test_custom_json_parser_invalid_json():
    parser = CustomJsonParser(pydantic_object=AnswerAndSources)
    llm_output = '```jsonn{"answer": "LangChain是一个框架。", "source_docs": ["文档A"]}n```' # 字段名错误

    with pytest.raises(Exception): # LangChain_core.output_parsers.json.JsonOutputParserError
        parser.parse(llm_output)

def test_custom_json_parser_no_json_block():
    parser = CustomJsonParser(pydantic_object=AnswerAndSources)
    llm_output = '这是一个普通文本，没有JSON。'

    with pytest.raises(Exception): # LangChain_core.output_parsers.json.JsonOutputParserError
        parser.parse(llm_output)

4.2 集成测试：验证链式逻辑与语义一致性

集成测试是回归测试的重中之重，它关注 LangChain 组件（如 Prompt、LLM、Retriever、Parser）组合成完整链条后的行为。这里的核心挑战是评估 LLM 输出的语义一致性。

我们将使用 LangChain 的 LCEL (LangChain Expression Language) 来构建链，因为它更具模块化和可测试性。

4.2.1 简单的问答链 (QA Chain) 测试

对比新旧版本链在给定输入下，输出是否在语义上保持一致。

# app/chains.py
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

def create_qa_chain(model_name: str = "gpt-3.5-turbo"):
    llm = ChatOpenAI(model_name=model_name, temperature=0) # 保持低温度以减少随机性

    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template("你是一个友好的AI助手，请简洁地回答问题。"),
        HumanMessagePromptTemplate.from_template("{question}")
    ])

    chain = prompt | llm | StrOutputParser()
    return chain

# tests/test_chains.py
import pytest
import json
from app.chains import create_qa_chain
from langchain_openai import OpenAIEmbeddings
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import os

# 确保环境变量设置
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") 

# 加载基线数据
def load_golden_data(filename):
    with open(filename, 'r', encoding='utf-8') as f:
        return json.load(f)

# 语义相似度评估函数
embeddings_model = OpenAIEmbeddings(model="text-embedding-ada-002")

def get_embedding(text: str) -> np.ndarray:
    return embeddings_model.embed_query(text)

def semantic_similarity(text1: str, text2: str) -> float:
    if not text1 or not text2: # 处理空字符串情况
        return 0.0
    emb1 = get_embedding(text1)
    emb2 = get_embedding(text2)
    return cosine_similarity([emb1], [emb2])[0][0]

# --- 测试用例 ---
@pytest.fixture(scope="module")
def qa_golden_data():
    # 假设这是旧版本（v0.0.350）生成的基线数据
    return load_golden_data("tests/golden_data/qa_chain_golden_v0_0_350.json")

def test_qa_chain_semantic_consistency(qa_golden_data):
    qa_chain = create_qa_chain()

    similarity_threshold = 0.85 # 定义语义相似度阈值

    for item in qa_golden_data:
        input_question = item["input"]
        expected_output = item["expected_output"]

        print(f"nTesting question: {input_question}")
        print(f"Expected (old version): {expected_output}")

        # 运行新版本的链
        actual_output = qa_chain.invoke({"question": input_question})
        print(f"Actual (new version): {actual_output}")

        # 1. 关键词检查 (简单判断)
        if "LangChain" in input_question:
            assert "LangChain" in actual_output or "框架" in actual_output, 
                f"Output for '{input_question}' missing key terms."

        # 2. 语义相似度检查 (更鲁棒)
        sim = semantic_similarity(expected_output, actual_output)
        print(f"Semantic similarity: {sim:.2f}")
        assert sim >= similarity_threshold, 
            f"Semantic shift detected for '{input_question}'. Similarity: {sim:.2f} < {similarity_threshold:.2f}"

        # 3. 长度变化检查 (可选，防止过于简短或冗长)
        # assert 0.7 < len(actual_output) / len(expected_output) < 1.3, 
        #     f"Output length changed significantly for '{input_question}'"

    # 模拟 golden_data 文件内容 (tests/golden_data/qa_chain_golden_v0_0_350.json)
    # [
    #     {"input": "LangChain是什么？", "expected_output": "LangChain是一个开源框架，用于开发由大型语言模型驱动的应用程序。"},
    #     {"input": "请解释一下人工智能。", "expected_output": "人工智能是一门研究如何使机器像人类一样思考、学习和行动的科学与技术。"},
    #     {"input": "谁是阿尔伯特·爱因斯坦？", "expected_output": "阿尔伯特·爱因斯坦是20世纪最伟大的物理学家之一，提出了相对论。"}
    # ]

说明：

• 温度设置： 将 temperature 设置为0，尽可能减少LLM的随机性，使输出更稳定。
• 基线数据： qa_golden_data 存储了旧版本在特定输入下的预期输出。
• 语义相似度： 这是一个强大的工具。我们使用 OpenAIEmbeddings 将文本转换为向量，然后计算余弦相似度。这比简单的字符串匹配更能捕捉语义上的变化。
• 阈值： similarity_threshold 定义了可接受的语义相似度下限。这个值需要根据实际应用场景进行调整和实验。
• 多重检查： 结合关键词检查和语义相似度，增加测试的鲁棒性。

4.2.2 RAG (Retrieval Augmented Generation) 链测试

对于 RAG 应用，除了生成内容的语义一致性，还需要关注检索到的文档的相关性。

# app/rag_chain.py
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document

def create_rag_chain(docs: list[str], model_name: str = "gpt-3.5-turbo"):
    llm = ChatOpenAI(model_name=model_name, temperature=0)
    embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")

    # 1. 文档处理与向量存储
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    documents = [Document(page_content=d) for d in docs]
    splits = text_splitter.split_documents(documents)
    vectorstore = FAISS.from_documents(splits, embeddings)
    retriever = vectorstore.as_retriever()

    # 2. Prompt模板
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个知识渊博的AI助手。请根据以下提供的上下文信息回答问题。如果上下文没有提供足够的信息，请说明你不知道。"),
        ("human", "上下文：{context}nn问题：{question}")
    ])

    # 3. 构建RAG链
    def format_docs(docs_list):
        return "nn".join(doc.page_content for doc in docs_list)

    rag_chain = (
        {"context": retriever | RunnableLambda(format_docs), "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    return rag_chain

# tests/test_rag_chain.py
import pytest
import json
from app.rag_chain import create_rag_chain
from langchain_openai import OpenAIEmbeddings
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import os

# 确保环境变量设置
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") 

# 语义相似度评估函数 (同上)
embeddings_model = OpenAIEmbeddings(model="text-embedding-ada-002")

def get_embedding(text: str) -> np.ndarray:
    return embeddings_model.embed_query(text)

def semantic_similarity(text1: str, text2: str) -> float:
    if not text1 or not text2:
        return 0.0
    emb1 = get_embedding(text1)
    emb2 = get_embedding(text2)
    return cosine_similarity([emb1], [emb2])[0][0]

# 模拟知识库文档
KNOWLEDGE_BASE_DOCS = [
    "LangChain是一个用于开发由大型语言模型驱动的应用程序的框架。它简化了LLM应用的开发过程。",
    "LangChain提供了各种模块，包括模型I/O、Prompt模板、链、Agent和检索器等。",
    "RAG（Retrieval Augmented Generation）是一种结合信息检索和文本生成的AI技术，旨在提高LLM生成答案的准确性和相关性。",
    "FAISS是一个高效的相似性搜索库，常用于向量存储。",
    "OpenAI是领先的AI研究公司，提供GPT系列模型和嵌入模型API。"
]

# 加载基线数据 (同上)
def load_golden_data(filename):
    with open(filename, 'r', encoding='utf-8') as f:
        return json.load(f)

@pytest.fixture(scope="module")
def rag_golden_data():
    # 假设这是旧版本生成的基线数据，包含生成的答案和检索到的文档片段
    return load_golden_data("tests/golden_data/rag_chain_golden_v0_0_350.json")

def test_rag_chain_consistency(rag_golden_data):
    rag_chain = create_rag_chain(KNOWLEDGE_BASE_DOCS)

    answer_similarity_threshold = 0.80
    retrieval_relevance_threshold = 0.60 # 检索到的文档与问题本身的相似度

    for item in rag_golden_data:
        question = item["question"]
        expected_answer = item["expected_answer"]
        expected_retrieved_docs_keywords = item.get("expected_retrieved_docs_keywords", []) # 期望检索到的文档中的关键词

        print(f"nTesting RAG for question: {question}")
        print(f"Expected Answer (old version): {expected_answer}")

        # 运行新版本的RAG链
        # 为了测试检索结果，我们需要分解链或者使用Callback
        # 这里我们简化，直接调用整个链，并通过LLM-as-a-Judge或人工评估检索质量

        # 实际运行链，获取最终答案
        actual_answer = rag_chain.invoke(question)
        print(f"Actual Answer (new version): {actual_answer}")

        # 1. 答案语义一致性检查
        answer_sim = semantic_similarity(expected_answer, actual_answer)
        print(f"Answer semantic similarity: {answer_sim:.2f}")
        assert answer_sim >= answer_similarity_threshold, 
            f"Answer semantic shift detected for '{question}'. Similarity: {answer_sim:.2f} < {answer_similarity_threshold:.2f}"

        # 2. 检索相关性检查 (更复杂，需要获取中间检索结果)
        # LangChain 0.1.x 及其 LCEL 提供了 `with_config({"callbacks": [callback_handler]})` 来获取中间结果
        # 假设我们能通过某种方式（例如自定义回调）捕获检索到的文档
        # 这里我们假设 `rag_chain` 可以返回一个包含 `context` 和 `answer` 的字典
        # 实际操作中，你可能需要修改 `create_rag_chain` 或者使用 `with_config` 和 `get_chain_context` 等方法来获取中间的检索结果。

        # 简化示例：假设我们无法直接获取检索结果，但可以通过答案反推
        # 更好的方法是使用 `RunnablePassthrough.assign()` 配合 `with_config` 和 `RunnableConfig` 来捕获
        # 例如：
        # def get_retrieved_docs(question_input):
        #     # 这是一个模拟，实际应该从链的中间步骤获取
        #     retriever = create_rag_chain(KNOWLEDGE_BASE_DOCS).get_retriever() # 假设可以获取
        #     return retriever.invoke(question_input)
        # retrieved_docs = get_retrieved_docs(question)
        # for kw in expected_retrieved_docs_keywords:
        #     assert any(kw in doc.page_content for doc in retrieved_docs), 
        #         f"Retrieved docs for '{question}' missing keyword: {kw}"

        # 3. LLM-as-a-Judge (使用另一个LLM评估)
        # 这是一个更高级的技巧，尤其适用于主观性强的评估
        # judge_llm = ChatOpenAI(model="gpt-4", temperature=0) # 使用更强大的模型做评判
        # evaluation_prompt = ChatPromptTemplate.from_messages([
        #     ("system", "你是一个公正的评判者。请评估两个答案的语义相似度，并给出一个0-100的评分。"),
        #     ("human", "原始答案：{original_answer}n新答案：{new_answer}n请给出相似度评分：")
        # ])
        # judge_chain = evaluation_prompt | judge_llm | StrOutputParser()
        # judge_result = judge_chain.invoke({"original_answer": expected_answer, "new_answer": actual_answer})
        # print(f"LLM Judge Score: {judge_result}")
        # assert int(judge_result) >= 70 # 假设70分以上算合格

    # 模拟 golden_data 文件内容 (tests/golden_data/rag_chain_golden_v0_0_350.json)
    # [
    #     {
    #         "question": "LangChain主要用来做什么？",
    #         "expected_answer": "LangChain主要用于开发由大型语言模型驱动的应用程序，它简化了LLM应用的构建过程。",
    #         "expected_retrieved_docs_keywords": ["LangChain", "框架", "LLM应用"]
    #     },
    #     {
    #         "question": "什么是RAG？",
    #         "expected_answer": "RAG（Retrieval Augmented Generation）是一种结合信息检索和文本生成的AI技术，旨在提高LLM生成答案的准确性和相关性。",
    #         "expected_retrieved_docs_keywords": ["RAG", "检索", "生成"]
    #     }
    # ]

说明：

• RAG的复杂性： RAG 测试不仅要看最终答案，还要看中间的检索结果。在 LCEL 中，可以通过 with_config 和自定义回调来捕获中间步骤。
• RunnablePassthrough 的作用： 在 LCEL 中，RunnablePassthrough 允许我们将输入原样传递给链的下一个组件，或者将其作为字典中的一个键传递。RunnableLambda 用于对输入或中间结果进行函数式处理。
• LLM-as-a-Judge： 当人工评估成本过高，且语义相似度不足以完全捕捉细微差别时，可以引入一个更强大的LLM（如GPT-4）作为“评委”来评估新旧输出的质量、相关性或一致性。这是一种非常有效的半自动化方法，但会增加API成本。

4.2.3 Agent 行为测试

Agent 的行为测试是最复杂的，因为它涉及多步骤推理、工具选择和动态交互。

# app/agent_chain.py
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
import os

# 示例工具
@tool
def multiply(a: int, b: int) -> int:
    """将两个数字相乘。"""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """将两个数字相加。"""
    return a + b

def create_math_agent(model_name: str = "gpt-3.5-turbo"):
    llm = ChatOpenAI(model_name=model_name, temperature=0)
    tools = [multiply, add]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个数学助手，可以使用工具进行计算。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}") # 必须包含此占位符
    ])

    agent = create_openai_tools_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    return agent_executor

# tests/test_agent_chain.py
import pytest
import json
from app.agent_chain import create_math_agent
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any

# 确保环境变量设置
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") 

# 自定义回调处理器，用于捕获Agent的中间步骤
class AgentStepCollector(BaseCallbackHandler):
    def __init__(self):
        self.steps = []
        self.final_output = None

    def on_agent_action(self, action: Any, **kwargs: Any) -> Any:
        self.steps.append({"type": "action", "tool": action.tool, "tool_input": action.tool_input})

    def on_tool_end(self, output: str, **kwargs: Any) -> Any:
        self.steps.append({"type": "tool_output", "output": output})

    def on_agent_finish(self, finish: Any, **kwargs: Any) -> Any:
        self.final_output = finish.return_values["output"]
        self.steps.append({"type": "finish", "output": self.final_output})

# 加载基线数据
def load_golden_data(filename):
    with open(filename, 'r', encoding='utf-8') as f:
        return json.load(f)

@pytest.fixture(scope="module")
def agent_golden_data():
    return load_golden_data("tests/golden_data/agent_golden_v0_0_350.json")

def test_math_agent_consistency(agent_golden_data):
    agent_executor = create_math_agent()

    for item in agent_golden_data:
        question = item["question"]
        expected_final_answer = item["expected_final_answer"]
        expected_steps = item.get("expected_steps", []) # 期望的Agent推理步骤

        print(f"nTesting Agent for question: {question}")
        print(f"Expected Final Answer (old version): {expected_final_answer}")

        collector = AgentStepCollector()

        # 运行新版本的Agent，并使用回调捕获步骤
        actual_output = agent_executor.invoke({"input": question}, config={"callbacks": [collector]})
        actual_final_answer = actual_output["output"]
        actual_steps = collector.steps

        print(f"Actual Final Answer (new version): {actual_final_answer}")
        print(f"Actual Steps: {actual_steps}")

        # 1. 最终答案检查
        assert actual_final_answer == expected_final_answer, 
            f"Agent final answer changed for '{question}'. Expected: {expected_final_answer}, Got: {actual_final_answer}"

        # 2. 步骤序列检查 (更严格的语义偏移检查)
        # 比较工具选择和工具输出是否一致
        assert len(actual_steps) == len(expected_steps), 
            f"Agent step count changed for '{question}'. Expected {len(expected_steps)}, Got {len(actual_steps)}"

        for i, (expected_step, actual_step) in enumerate(zip(expected_steps, actual_steps)):
            assert expected_step["type"] == actual_step["type"], 
                f"Step {i} type mismatch for '{question}'."

            if expected_step["type"] == "action":
                assert expected_step["tool"] == actual_step["tool"], 
                    f"Step {i} tool mismatch for '{question}'."
                # 进一步检查 tool_input，可能需要进行模糊匹配或语义相似度
                assert expected_step["tool_input"] == actual_step["tool_input"], 
                    f"Step {i} tool input mismatch for '{question}'."
            elif expected_step["type"] == "tool_output":
                # 工具输出可能略有变化，尤其是有外部依赖时
                assert expected_step["output"] == actual_step["output"], 
                    f"Step {i} tool output mismatch for '{question}'."
            elif expected_step["type"] == "finish":
                assert expected_step["output"] == actual_step["output"], 
                    f"Step {i} final output mismatch for '{question}'."

    # 模拟 golden_data 文件内容 (tests/golden_data/agent_golden_v0_0_350.json)
    # [
    #     {
    #         "question": "计算 (5 + 3) * 2",
    #         "expected_final_answer": "16",
    #         "expected_steps": [
    #             {"type": "action", "tool": "add", "tool_input": {"a": 5, "b": 3}},
    #             {"type": "tool_output", "output": "8"},
    #             {"type": "action", "tool": "multiply", "tool_input": {"a": 8, "b": 2}},
    #             {"type": "tool_output", "output": "16"},
    #             {"type": "finish", "output": "16"}
    #         ]
    #     },
    #     {
    #         "question": "请将 10 乘以 5",
    #         "expected_final_answer": "50",
    #         "expected_steps": [
    #             {"type": "action", "tool": "multiply", "tool_input": {"a": 10, "b": 5}},
    #             {"type": "tool_output", "output": "50"},
    #             {"type": "finish", "output": "50"}
    #         ]
    #     }
    # ]

说明：

• AgentStepCollector： 这是一个关键的回调处理器，用于捕获 Agent 在执行过程中采取的每一个行动（调用哪个工具、输入是什么）和工具的输出。这使得我们可以对 Agent 的推理路径进行细粒度验证。
• 基线步骤： expected_steps 存储了旧版本 Agent 在处理相同问题时所采取的精确步骤序列。
• 严格比较： 对于 Agent 的工具选择和输入，我们倾向于进行更严格的比较，因为这些是 Agent 决策的核心。工具输出可能因外部因素略有不同，但其语义应该一致。

4.3 端到端测试：模拟真实用户场景

端到端测试模拟用户与整个应用（包括前端/API层）的交互。这通常涉及到更高级的测试框架（如 Selenium for Web UI, requests for API），但其核心仍然是验证 LangChain 后端服务的行为。

在 LangChain 语境下，一个完整的端到端测试可能意味着：

1. 通过一个 REST API 调用你的 LangChain 服务。
2. 模拟一个聊天机器人界面，发送消息并接收回复。

由于篇幅限制，这里只提供一个概念性的 API 端到端测试框架示例。

# app/main.py (假设你的LangChain应用通过FastAPI暴露API)
from fastapi import FastAPI
from pydantic import BaseModel
from app.agent_chain import create_math_agent
import os

os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR_OPENAI_API_KEY") 

app = FastAPI()
math_agent = create_math_agent()

class QueryRequest(BaseModel):
    question: str

class QueryResponse(BaseModel):
    answer: str

@app.post("/query", response_model=QueryResponse)
async def query_agent(request: QueryRequest):
    result = await math_agent.ainvoke({"input": request.question})
    return QueryResponse(answer=result["output"])

# tests/test_e2e_api.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
import json

# 创建测试客户端
client = TestClient(app)

# 加载基线数据
def load_golden_data(filename):
    with open(filename, 'r', encoding='utf-8') as f:
        return json.load(f)

@pytest.fixture(scope="module")
def api_golden_data():
    return load_golden_data("tests/golden_data/e2e_api_golden_v0_0_350.json")

def test_api_math_agent_e2e(api_golden_data):
    for item in api_golden_data:
        question = item["question"]
        expected_answer = item["expected_answer"]

        print(f"nTesting API E2E for question: {question}")
        print(f"Expected Answer (old version): {expected_answer}")

        response = client.post("/query", json={"question": question})
        assert response.status_code == 200

        actual_response_data = response.json()
        actual_answer = actual_response_data["answer"]

        print(f"Actual Answer (new version): {actual_answer}")

        assert actual_answer == expected_answer, 
            f"E2E API final answer changed for '{question}'. Expected: {expected_answer}, Got: {actual_answer}"

    # 模拟 golden_data 文件内容 (tests/golden_data/e2e_api_golden_v0_0_350.json)
    # [
    #     {"question": "计算 (10 + 2) * 3", "expected_answer": "36"},
    #     {"question": "请告诉我 7 乘以 8 的结果是多少？", "expected_answer": "56"}
    # ]

说明：

• FastAPI TestClient： 这是一个轻量级的工具，用于测试 FastAPI 应用的 HTTP 端点，而无需实际启动服务器。
• 覆盖范围： 端到端测试验证的是从用户请求到最终响应的整个流程，包括数据序列化/反序列化、路由、LangChain 后端逻辑等。
• 与集成测试的协同： 端到端测试可以作为集成测试的补充，验证最外层的接口是否正常，但对于内部的细致语义偏移，集成测试通常提供更精确的定位能力。

---

第五讲：LLM测试的工具与框架

除了标准的 pytest 和 unittest，业界也涌现出一些专门针对LLM应用测试的工具和框架。

1. LangChain 自身的测试模块 (langchain.testing):

   • LangChain 官方也在构建其测试工具。例如，langchain_core.tracers 模块可以用于跟踪和记录链的执行过程，这对于捕获 Agent 步骤和中间结果非常有用。
   • 未来可能会有更成熟的、专门用于比较不同版本链行为的工具。
   • 建议： 关注官方文档和更新日志，利用其最新提供的测试能力。

2. Promptfoo:

   • 一个开源的CLI工具，用于测试和评估 LLM prompts、chains 和 RAG 应用。
   • 支持多种 LLM 提供商，可以并行运行大量测试用例。
   • 提供可视化界面来比较不同版本或不同模型下的输出，可以进行手动评分、关键词检查、正则表达式匹配等。
   • 非常适合进行大规模的基线测试和比较。

   # promptfoo.yaml 示例
   prompts:
     - "你是一个友好的AI助手，请简洁地回答问题。n问题：{{question}}" # LangChain Prompt模板
   providers:
     - id: openai:chat:gpt-3.5-turbo # 可以指定不同的模型或版本
       config:
         temperature: 0
   tests:
     - vars:
         question: "LangChain是什么？"
       assert:
         - type: similar
           value: "LangChain是一个用于开发由大型语言模型驱动的应用程序的框架。"
           threshold: 0.8
     - vars:
         question: "请解释一下人工智能。"
       assert:
         - type: regex
           value: "人工智能.*科学与技术"
         - type: llm-rubric
           value: "答案是否准确且简洁？" # LLM-as-a-Judge

3. DeepEval:

   • 一个 Python 库，专注于 LLM 评估和测试。
   • 提供了多种内置指标（如 AnswerRelevancy, Faithfulness, ContextualRecall 等），可以直接用于评估 RAG 链的质量。
   • 支持自定义评估器，可以集成到 CI/CD 流程中。

   # DeepEval 示例
   from deepeval.test_case import LLMTestCase
   from deepeval.metrics import AnswerRelevancyMetric
   from deepeval.run_test import run_test
   
   # 假设你的RAG链
   def my_rag_chain(question: str) -> str:
       # ... 调用你的LangChain RAG链
       return "LangChain是一个用于开发由大型语言模型驱动的应用程序的框架。"
   
   test_case = LLMTestCase(
       input="LangChain是什么？",
       actual_output=my_rag_chain("LangChain是什么？"),
       retrieval_context=["LangChain是一个开源框架。", "它简化了LLM应用的开发过程。"], # 如果可以获取检索上下文
       expected_output="LangChain是一个用于开发由大型语言模型驱动的应用程序的框架。"
   )
   
   metric = AnswerRelevancyMetric(threshold=0.7)
   run_test([test_case], metrics=[metric])

4. 自定义脚本和工具：

   • 对于高度定制化的需求，结合 pytest、OpenAIEmbeddings 和一些文本处理库（如 nltk, spacy）构建自己的比较和评估脚本，仍然是非常灵活和强大的方法。

---

第六讲：LangChain升级与回归测试的工作流

将上述原则和实践整合到一个完整的工作流中，以指导实际的 LangChain 升级过程。

1. 准备阶段（Pre-Upgrade）：

   • 备份代码和环境： 确保当前工作版本稳定，并进行代码备份或提交到版本控制。
   • 冻结依赖： 运行 pip freeze > requirements_old.txt，记录当前所有库的版本。
   • 建立基线数据集（Golden Dataset）：
     • 识别关键业务流程和高风险组件。
     • 为这些流程和组件创建全面的输入数据集。
     • 在当前（旧）版本的应用上运行这些输入，并保存输出作为基线。这包括最终答案、Agent步骤、检索文档等关键中间结果。
     • 尽可能使用自动化脚本生成基线。
   • 确保现有测试通过： 运行所有已有的单元、集成、端到端测试，确保旧版本是健康的。

2. 升级阶段（Upgrade）：

   • 创建新分支： 在版本控制系统中创建一个专门用于升级的新分支（例如 feature/langchain-v0.1-upgrade）。
   • 更新 requirements.txt： 修改 LangChain 及其相关依赖的版本号。
   • 安装新依赖： 在新的虚拟环境中运行 pip install -r requirements.txt。
   • 修复破坏性变更： 根据 LangChain 的升级日志和错误信息，修改代码以兼容新的 API。这一步通常是显性的。

3. 测试阶段（Testing）：

   • 运行单元测试： 确保自定义组件（Prompt、工具、解析器）在新版本下依然正常工作。
   • 运行集成测试： 这是核心。
     • 使用自动化测试脚本，加载预先建立的基线数据。
     • 在新版本应用上运行相同的输入。
     • 比较新旧版本的输出，评估语义相似度、准确性、Agent行为等，并根据预设的阈值判断是否通过。
     • 关注性能指标（延迟、成本）是否有显著变化。
   • 运行端到端测试： 验证整个应用流程是否顺畅，用户体验是否受损。
   • 人工评审（Human-in-the-Loop）： 对于通过自动化测试但仍有疑虑，或涉及高度主观判断的输出，进行人工抽样评审。特别是对于Agent的复杂推理路径，人工检查其思考过程至关重要。

4. 分析与迭代（Analyze & Iterate）：

   • 识别回归： 如果测试失败或发现语义偏移，分析原因。是 LangChain 内部逻辑变化？是 Prompt 渲染差异？是底层 LLM 模型行为改变？
   • 调整代码： 根据分析结果，调整你的 LangChain 代码（例如，更新 Prompt 策略、调整链的配置、优化解析器）。
   • 更新基线（如果必要）： 如果确定新版本的行为是“更好”或“可接受”的新标准，并且不希望回滚，那么需要更新部分基线数据以反映新的预期。这需要谨慎操作，并记录原因。
   • 重复测试： 修复后，重新运行所有受影响的测试。

5. 部署阶段（Deployment）：

   • 合入主分支： 确认所有测试通过，且回归风险可控后，将升级分支合入主分支。
   • 监控： 在生产环境中持续监控应用表现，包括错误率、性能、用户反馈等，及时发现潜在的、未被测试捕获的问题。

---

确保应用持续智能的基石

在LLM技术飞速发展的今天，LangChain等框架的迭代是必然的，也是我们拥抱新技术、提升应用能力的关键。但每一次升级，都伴随着潜在的语义偏移风险。回归测试，特别是针对LLM应用特性的多维度、分层、自动化与人工结合的回归测试，是我们确保应用稳定、可靠，并持续提供高价值智能服务的基石。它不仅仅是发现和修复bug的手段，更是我们理解和管理LLM非确定性行为，提升产品质量信心的重要实践。投资于健壮的测试策略，就是投资于你的AI应用的未来。