第十篇 生产实践与监控评估#

root_run (Agent 执行)
├── before_model_hook (Middleware)
├── model_call (模型调用)
│   ├── system_prompt
│   ├── messages
│   └── tools
├── tool_run (工具执行)
│   ├── search_tool
│   └── get_weather_tool
└── after_model_hook (后处理)

class Run:
    id: str                          # 唯一 ID
    name: str                        # 运行名称
    run_type: str                    # "agent", "model", "tool", "chain" 等
    parent_run_id: Optional[str]     # 父 Run ID（形成树关系）

    # 输入输出
    inputs: dict[str, Any]           # 输入参数
    outputs: dict[str, Any]          # 输出结果

    # 时间和成本
    start_time: datetime             # 开始时间
    end_time: datetime               # 结束时间
    duration: float                  # 执行耗时（秒）

    # Token 和成本
    token_usage: Optional[TokenUsage]
    cost: Optional[float]            # 美元成本

    # 状态和错误
    status: str                      # "success", "error"
    error: Optional[str]             # 错误信息

    # 元数据
    metadata: dict[str, Any]         # 自定义元数据
    tags: list[str]                  # 标签（用于筛选）

    # 反馈
    feedback_records: list[Feedback] # 用户反馈

# 设置环境变量（LangSmith 1.0+ 最新命名）
export LANGSMITH_API_KEY="lsv2_..."
export LANGCHAIN_PROJECT="my_project"
export LANGSMITH_TRACING="true"

import os
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent

# 配置 LangSmith（最新命名）
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGCHAIN_PROJECT"] = "my_project"
os.environ["LANGSMITH_TRACING"] = "true"

# 创建 Agent
model = ChatOpenAI(model="gpt-4o")
agent = create_agent(
    model=model,
    tools=[search, weather],
    system_prompt="你是一个助手"
)

# ✅ 自动追踪：所有调用都会被记录到 LangSmith
result = agent.invoke({"messages": [("user", "今天天气如何")]})

# 查看追踪：
# 1. 打开 https://smith.langchain.com
# 2. 选择项目 "my_project"
# 3. 查看实时追踪树

import os

# 基础配置（必需）- LangSmith 1.0+ 最新命名
os.environ["LANGSMITH_API_KEY"] = "lsv2_..."
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my_project"

# 团队协作配置（可选）
os.environ["LANGSMITH_WORKSPACE_ID"] = "ws-..."  # 工作区 ID（团队共享）

# 可选配置
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"  # 自定义端点
os.environ["LANGCHAIN_SESSION"] = "user-session-123"  # 会话分组
os.environ["LANGCHAIN_CALLBACKS_BACKGROUND"] = "true"  # 后台发送，不阻塞主程序
os.environ["LANGCHAIN_TRACING_SAMPLING_RATE"] = "0.1"  # 生产环境采样 10%

# 开发/测试环境配置
os.environ["LANGSMITH_TEST_CACHE"] = "true"  # 缓存 API 调用以加速测试

# 向后兼容（不推荐，仍然支持）
# os.environ["LANGCHAIN_API_KEY"] = "sk-..."  # 已弃用，使用 LANGSMITH_API_KEY
# os.environ["LANGCHAIN_TRACING_V2"] = "true"  # 已弃用，使用 LANGSMITH_TRACING

from langsmith.run_helpers import traceable

@traceable(name="my_function", run_type="chain")
def my_custom_function(input_text: str) -> str:
    """自定义函数会自动被追踪"""
    # 处理输入
    result = input_text.upper()
    return result

# 调用时自动记录到 LangSmith
output = my_custom_function("hello")

from langsmith import traceable

@traceable(
    # 基础参数
    name="custom_processing",              # 运行名称（默认：函数名）
    run_type="tool",                       # 运行类型：llm/chain/tool/retriever/prompt（默认：chain）

    # 组织和标记
    metadata={                             # 元数据（任意键值对）
        "version": "1.0",
        "author": "alice",
        "module": "data_processing"
    },
    tags=["production", "v1"],             # 标签列表（用于筛选）
    project_name="my_project",             # 项目名称（覆盖环境变量）

    # 高级参数
    reduce_fn=None,                        # 聚合函数（用于生成器/流式输出）
    client=None,                           # 自定义 LangSmith Client
    process_inputs=None,                   # 输入预处理函数
    process_outputs=None,                  # 输出后处理函数
)
def process_data(data: dict) -> dict:
    """处理数据"""
    result = {**data, "processed": True}
    return result

# 1. 处理生成器输出
@traceable(
    name="stream_processor",
    reduce_fn=lambda outputs: "".join(outputs)  # 将流式输出合并为字符串
)
def stream_data(n: int):
    """生成器函数"""
    for i in range(n):
        yield f"chunk-{i}"

# 2. 自定义输入/输出处理
def serialize_inputs(inputs):
    """自定义输入序列化"""
    return {k: str(v)[:100] for k, v in inputs.items()}  # 截断长文本

@traceable(
    process_inputs=serialize_inputs,
    process_outputs=lambda o: {"result": str(o)[:200]}
)
def process_large_data(data: dict) -> dict:
    # 处理大量数据，但只记录摘要
    return {"result": data}

@traceable(name="parent_task")
def parent_task(query: str) -> str:
    # 调用子任务
    result1 = search_task(query)
    result2 = analyze_task(result1)
    return result2

@traceable(name="search_task")
def search_task(query: str) -> str:
    # 子任务 1
    return f"搜索结果: {query}"

@traceable(name="analyze_task")
def analyze_task(text: str) -> str:
    # 子任务 2
    return f"分析: {text}"

# 调用时自动形成树：
# parent_task
# ├── search_task
# └── analyze_task
parent_task("LangChain")

from langsmith.run_helpers import traceable, get_current_run_tree

@traceable
def my_function(x: int) -> int:
    # 获取当前运行
    run = get_current_run_tree()

    if run:
        print(f"当前运行 ID: {run.id}")
        print(f"运行名称: {run.name}")

        # 添加自定义元数据
        run.metadata = {"user_id": "user-123"}

    return x * 2

import asyncio
from langsmith import traceable
import httpx

@traceable(name="fetch_data", run_type="retriever")
async def fetch_user_data(user_id: str) -> dict:
    """异步获取用户数据"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

@traceable(name="process_user", run_type="chain")
async def process_user(user_id: str) -> dict:
    """异步处理用户信息"""
    # 异步调用会自动追踪
    user_data = await fetch_user_data(user_id)

    # 处理数据
    processed = {
        "id": user_data["id"],
        "name": user_data["name"].upper(),
        "processed_at": "2025-01-15"
    }

    return processed

# 使用异步追踪
async def main():
    result = await process_user("user-123")
    print(result)

# 执行
asyncio.run(main())

# 追踪树会显示：
# process_user (async)
# └── fetch_user_data (async)

@traceable(name="batch_fetch", run_type="chain")
async def fetch_multiple_users(user_ids: list[str]) -> list[dict]:
    """并发获取多个用户数据"""
    # 所有异步调用都会被追踪
    tasks = [fetch_user_data(uid) for uid in user_ids]
    results = await asyncio.gather(*tasks)
    return results

# 使用
async def main():
    users = await fetch_multiple_users(["user-1", "user-2", "user-3"])

# 追踪树会显示：
# batch_fetch
# ├── fetch_user_data (user-1)
# ├── fetch_user_data (user-2)
# └── fetch_user_data (user-3)

@traceable(name="safe_operation", run_type="tool")
def safe_operation(value: int) -> int:
    """带错误处理的操作"""
    try:
        if value < 0:
            raise ValueError("Value must be non-negative")

        result = 100 / value
        return result

    except ValueError as e:
        # 异常会自动记录到 LangSmith
        print(f"Validation error: {e}")
        raise  # 重新抛出以标记为失败

    except ZeroDivisionError as e:
        # 除零错误也会被记录
        print(f"Division error: {e}")
        raise

# 调用失败的函数
try:
    safe_operation(0)  # 会在 LangSmith 中标记为 error
except ZeroDivisionError:
    pass

# LangSmith 中会显示：
# - 运行状态: error
# - 错误类型: ZeroDivisionError
# - 错误消息: division by zero
# - 完整堆栈跟踪

@traceable(name="async_safe_fetch")
async def async_safe_fetch(url: str) -> dict:
    """带错误处理的异步请求"""
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url, timeout=5.0)
            response.raise_for_status()
            return response.json()

    except httpx.TimeoutException as e:
        # 超时错误会被记录
        print(f"Request timeout: {e}")
        raise

    except httpx.HTTPStatusError as e:
        # HTTP 错误会被记录
        print(f"HTTP error: {e.response.status_code}")
        raise

    except Exception as e:
        # 所有其他错误也会被记录
        print(f"Unexpected error: {e}")
        raise

# 使用
async def main():
    try:
        data = await async_safe_fetch("https://invalid-url.example.com")
    except Exception:
        # LangSmith 会记录完整的错误信息
        pass

from langsmith import Client

client = Client()

# 获取最近的追踪
runs = client.list_runs(
    project_name="my_project",
    limit=10  # 获取最近 10 条
)

for run in runs:
    print(f"运行: {run.name}")
    print(f"状态: {run.status}")
    print(f"耗时: {run.end_time - run.start_time}")

    if run.token_usage:
        print(f"Tokens: {run.token_usage.completion_tokens}")
        print(f"成本: ${run.cost}")

    # 获取子运行
    if run.child_runs:
        for child in run.child_runs:
            print(f"  └─ {child.name}: {child.status}")

from datetime import datetime, timedelta

# 筛选条件
yesterday = datetime.now() - timedelta(days=1)

# 获取特定标签的运行
production_runs = client.list_runs(
    project_name="my_project",
    filter='tags:production',
    start_time=yesterday
)

# 统计
total_cost = sum(run.cost for run in production_runs if run.cost)
avg_duration = sum(
    (run.end_time - run.start_time).total_seconds()
    for run in production_runs
) / len(production_runs)

error_count = sum(1 for run in production_runs if run.status == "error")

print(f"总成本: ${total_cost:.2f}")
print(f"平均耗时: {avg_duration:.2f} 秒")
print(f"错误数: {error_count}")

# 在应用中收集用户反馈
run_id = "run-123"  # 从追踪中获取

# 用户点赞
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=1.0,  # 1.0 = 好, 0.0 = 不好
    comment="很有用！"
)

# 专家标注
client.create_feedback(
    run_id=run_id,
    key="expert_evaluation",
    score=0.8,
    comment="大部分正确，但有一处错误"
)

# 查看反馈
feedbacks = client.list_feedbacks(run_id=run_id)
for feedback in feedbacks:
    print(f"{feedback.key}: {feedback.score} - {feedback.comment}")

from langsmith import Client

client = Client()

# 方法 1：手动创建
dataset = client.create_dataset(
    dataset_name="qa_benchmark",
    description="QA 任务基准数据集"
)

# 添加示例
client.create_example(
    dataset_id=dataset.id,
    inputs={"question": "LangChain 是什么？"},
    outputs={"answer": "LangChain 是一个构建 LLM 应用的框架"}
)

client.create_example(
    dataset_id=dataset.id,
    inputs={"question": "Python 3.13 的主要特性？"},
    outputs={"answer": "Python 3.13 引入了 JIT 编译器..."}
)

# 方法 2：从 CSV 导入
import pandas as pd

df = pd.read_csv("qa_examples.csv")
# 假设 CSV 有 "question" 和 "answer" 列

dataset = client.create_dataset(
    dataset_name="qa_from_csv",
    description="从 CSV 导入的 QA 数据集"
)

for _, row in df.iterrows():
    client.create_example(
        dataset_id=dataset.id,
        inputs={"question": row["question"]},
        outputs={"answer": row["answer"]}
    )

# 方法 3：从生产追踪创建（最推荐）
# 在生产环境运行一段时间后，筛选高质量的追踪
client.create_dataset(
    dataset_name="production_examples",
    description="从生产环境采集的真实用户查询"
)

# 手动筛选好的追踪并添加
# （通常通过 LangSmith UI 完成，支持批量导入）

# ✅ 好：覆盖多种场景
examples = [
    # 简单查询
    {"question": "天气如何？", "answer": "..."},

    # 复杂查询
    {"question": "比较 LangChain 和 Langraph 的优缺点", "answer": "..."},

    # 边界情况
    {"question": "", "answer": "请输入有效问题"},
    {"question": "???.|||", "answer": "请输入有效问题"},

    # 多轮对话
    {"question": "什么是 RAG？", "answer": "..."},
    {"question": "能给个例子吗？", "answer": "..."},
]

client.create_example(
    dataset_id=dataset.id,
    inputs={"question": "LangChain 是什么？"},
    outputs={"answer": "LangChain 是..."},
    metadata={
        "difficulty": "easy",
        "category": "framework",
        "source": "user_feedback",
        "language": "chinese"
    }
)

# 最简格式（只包含必需字段）
return {
    "key": "accuracy",
    "score": 1.0
}

# 完整格式（包含所有字段）
return {
    "key": "accuracy",           # 指标名称
    "score": 0.85,                # 分数（0-1 或任意数值）
    "value": 0.85,                # 实际值（可选，通常与 score 相同）
    "comment": "Good answer!",    # 评论（可选，用于解释）
    "correction": {               # 修正建议（可选）
        "expected": "Paris",
        "actual": "paris"
    }
}

# 布尔分数
return {
    "key": "is_correct",
    "score": True  # 布尔值也可以
}

# 多个评估结果（返回列表）
return [
    {"key": "accuracy", "score": 0.9},
    {"key": "latency", "score": 1.5, "comment": "1.5 seconds"}
]

def is_valid_code(run, example):
    """检查输出是否是有效的 Python 代码"""
    code = run.outputs.get("code", "")

    try:
        compile(code, "<string>", "exec")
        return {
            "key": "valid_code",
            "score": 1.0,
            "comment": "Valid Python code"
        }
    except SyntaxError as e:
        return {
            "key": "valid_code",
            "score": 0.0,
            "comment": f"Syntax error: {str(e)}",
            "correction": {"error": str(e)}
        }

def exact_match(run, example):
    """精确匹配评估器"""
    actual = run.outputs.get("answer")
    expected = example.outputs.get("answer")
    is_match = actual == expected

    return {
        "key": "exact_match",
        "score": 1.0 if is_match else 0.0,
        "value": is_match,
        "comment": "Match!" if is_match else f"Expected: {expected}, Got: {actual}",
        "correction": {"expected": expected} if not is_match else None
    }

def length_check(run, example):
    """检查输出长度"""
    answer = run.outputs.get("answer", "")
    length = len(answer)

    return {
        "key": "answer_length",
        "score": length,  # 分数可以是任意数值
        "value": length,
        "comment": f"Answer has {length} characters"
    }

from langchain_openai import ChatOpenAI
from langsmith.evaluation import LangChainStringEvaluator

# 使用官方提供的 LLM 评估器
from langsmith.evaluation import (
    EvaluateStrings,
    evaluate_strings
)

# 自定义 LLM 评估器
def llm_evaluator(run, example):
    """使用 LLM 评估相关性"""
    model = ChatOpenAI(model="gpt-4o-mini")

    evaluation_prompt = f"""
    问题：{example.inputs.get("question")}
    参考答案：{example.outputs.get("answer")}
    实际答案：{run.outputs.get("answer")}

    请评估实际答案与参考答案的相似度（0-1）。
    """

    response = model.invoke(evaluation_prompt)

    # 解析响应
    score = float(response.content.split("\n")[0])

    return {
        "score": score,
        "key": "relevance",
        "comment": response.content
    }

# LangSmith UI 中可以：
# 1. 在 Annotation Queue 中标注
# 2. 为每条结果打分或写评论
# 3. 导出评估结果供统计

from langsmith.evaluation import evaluate
from langsmith import Client

client = Client()

# 1. 使用数据集名称（最常用）
results = evaluate(
    predict,
    data="qa_benchmark",  # 数据集名称
    evaluators=[accuracy]
)

# 2. 使用数据集 UUID
results = evaluate(
    predict,
    data="a3d2f1b8-1234-5678-90ab-cdef12345678",  # UUID
    evaluators=[accuracy]
)

# 3. 直接传入示例列表（快速测试）
examples = [
    {
        "inputs": {"question": "What is LangChain?"},
        "outputs": {"answer": "LangChain is a framework..."}
    },
    {
        "inputs": {"question": "What is Python?"},
        "outputs": {"answer": "Python is a programming language..."}
    }
]
results = evaluate(
    predict,
    data=examples,  # 列表
    evaluators=[accuracy]
)

# 4. 使用迭代器（大数据集，节省内存）
def example_generator():
    """生成器函数，逐个产生示例"""
    for i in range(1000):
        yield {
            "inputs": {"question": f"Question {i}"},
            "outputs": {"answer": f"Answer {i}"}
        }

results = evaluate(
    predict,
    data=example_generator(),  # 迭代器
    evaluators=[accuracy]
)

# 5. 使用 LangSmith Example 对象迭代器
examples_iter = client.list_examples(
    dataset_name="qa_benchmark",
    limit=100  # 只测试前 100 个
)
results = evaluate(
    predict,
    data=examples_iter,
    evaluators=[accuracy]
)

from langsmith.evaluation import evaluate
from langsmith import Client

client = Client()

# 定义预测函数
def predict(input_dict):
    """应用的预测函数"""
    question = input_dict.get("question")

    # 调用 Agent
    agent = create_agent(model=ChatOpenAI(model="gpt-4o-mini"), tools=[...])
    result = agent.invoke({"messages": [("user", question)]})

    return {"answer": result["messages"][-1].content}

# 定义评估器
def exact_match(run, example):
    """精确匹配评估"""
    return {
        "score": 1.0 if run.outputs["answer"] == example.outputs["answer"] else 0.0,
        "key": "exact_match"
    }

def relevance(run, example):
    """相关性评估"""
    model = ChatOpenAI(model="gpt-4o-mini")
    response = model.invoke(
        f"问题: {example.inputs['question']}\n答案: {run.outputs['answer']}\n"
        f"评估相关性(0-1):"
    )
    score = float(response.content)
    return {"score": score, "key": "relevance"}

# 运行评估
experiment_results = evaluate(
    predict,                                    # 预测函数
    data="qa_benchmark",                       # 数据集名称
    evaluators=[exact_match, relevance],       # 行级评估器
    experiment_prefix="Model v1",              # 实验名称前缀
    description="测试新模型版本",
    num_repetitions=2,                         # 运行 2 次（处理 LLM 非确定性）
    max_concurrency=10                         # 并发数
)

# 查看结果
print(f"实验名: {experiment_results.experiment_name}")
print(f"总样例数: {len(experiment_results.results)}")

# 统计指标
scores = [r.evaluation_results[0].score for r in experiment_results.results]
print(f"平均准确率: {sum(scores) / len(scores):.2%}")

from langsmith.evaluation import evaluate

def accuracy_summary(runs, examples):
    """实验级评估器（在所有运行上计算）"""
    correct = 0
    for run, example in zip(runs, examples):
        if run.outputs["answer"] == example.outputs["answer"]:
            correct += 1

    accuracy = correct / len(runs)
    return {"score": accuracy, "key": "accuracy"}

def precision_recall_summary(runs, examples):
    """计算 Precision 和 Recall"""
    predictions = [run.outputs["answer"] for run in runs]
    references = [example.outputs["answer"] for example in examples]

    # 假设输出是分类标签列表
    tp = sum(1 for p, r in zip(predictions, references) if p == r and p == "positive")
    fp = sum(1 for p, r in zip(predictions, references) if p != r and p == "positive")
    fn = sum(1 for p, r in zip(predictions, references) if p != r and r == "positive")

    precision = tp / (tp + fp) if (tp + fp) > 0 else 0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0

    return [
        {"score": precision, "key": "precision"},
        {"score": recall, "key": "recall"}
    ]

# 使用 Summary Evaluator
evaluate(
    predict,
    data="classification_dataset",
    evaluators=[exact_match],
    summary_evaluators=[accuracy_summary, precision_recall_summary],
    experiment_prefix="Classification v1"
)

import asyncio
from langsmith.evaluation import aevaluate

async def main():
    # 异步预测函数
    async def async_predict(input_dict):
        # 异步 Agent 调用
        result = await async_agent.ainvoke({"messages": [("user", input_dict["question"])]})
        return {"answer": result["messages"][-1].content}

    # 运行异步评估
    results = await aevaluate(
        async_predict,
        data="large_dataset",
        evaluators=[exact_match, relevance],
        experiment_prefix="Async Model v1",
        max_concurrency=50  # 并发 50 个请求
    )

    return results

# 执行
results = asyncio.run(main())

from langsmith import Client

client = Client()

# 应用中收集用户反馈
run_id = "run-123"

# 用户点赞/点踩
client.create_feedback(
    run_id=run_id,
    key="user_rating",
    score=1.0,  # 1.0 = 好, 0.0 = 不好
    comment="回答很准确！"
)

# 专家标注
client.create_feedback(
    run_id=run_id,
    key="expert_annotation",
    score=0.8,
    comment="大部分正确，但第二点不够准确"
)

# 导出所有反馈用于分析
feedbacks = client.list_feedbacks(
    run_ids=[run_id]
)

for feedback in feedbacks:
    print(f"{feedback.key}: {feedback.score}")
    print(f"  {feedback.comment}")

from langsmith import Client
from langchain_core.prompts import ChatPromptTemplate

client = Client()

# 方法 1：将提示词推送到 Hub
system_prompt_v1 = """你是一个专业的编程助手。

能力：
1. 回答编程问题
2. 生成代码
3. 调试错误

约束：
- 只讨论编程相关话题
- 生成的代码要有注释
"""

# 创建提示词模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system", system_prompt_v1),
    ("user", "{input}")
])

# 推送到 Hub（需要 API 权限）
client.push_prompt(
    prompt_identifier="programming_assistant_system_prompt",
    object=prompt_template,
    description="编程助手的系统提示词"
)

# 方法 2：从 Hub 拉取提示词
system_prompt = client.pull_prompt("programming_assistant_system_prompt")

# 方法 3：版本管理
# 每次推送自动创建新版本，可以：
# 1. 查看历史版本
# 2. 对比版本差异
# 3. 拉取特定版本

updated_prompt_v2 = """你是一个专业的编程助手。

增强能力：
1. 回答编程问题
2. 生成代码
3. 调试错误

约束：
- 只讨论编程相关话题
- 生成的代码要有注释
"""

# 推送新版本（自动创建版本 2）
updated_template = ChatPromptTemplate.from_messages([
    ("system", updated_prompt_v2),
    ("user", "{input}")
])

client.push_prompt(
    prompt_identifier="programming_assistant_system_prompt",
    object=updated_template,
    description="添加 '增强' 前缀以强调能力"
)

# 在 LangSmith UI 中：
# 1. 打开 Prompt Hub
# 2. 选择提示词
# 3. 点击 "Compare Versions"
# 4. 查看变化和对比结果

from langsmith.evaluation import evaluate
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent

# 版本 A：原始模型和提示词
def predict_v_a(input_dict):
    agent = create_agent(
        model=ChatOpenAI(model="gpt-4o-mini"),
        tools=[...],
        system_prompt="你是一个助手"
    )
    result = agent.invoke({"messages": [("user", input_dict["question"])]})
    return {"answer": result["messages"][-1].content}

# 版本 B：改进的提示词
def predict_v_b(input_dict):
    agent = create_agent(
        model=ChatOpenAI(model="gpt-4o-mini"),
        tools=[...],
        system_prompt="""你是一个专业的助手。

在回答前：
1. 理解问题的关键信息
2. 分解复杂问题为子问题
3. 使用工具获取必要信息

最后提供清晰、准确的答案。"""
    )
    result = agent.invoke({"messages": [("user", input_dict["question"])]})
    return {"answer": result["messages"][-1].content}

# 定义评估器
def exact_match(run, example):
    return {
        "score": 1.0 if run.outputs["answer"] == example.outputs["answer"] else 0.0,
        "key": "exact_match"
    }

# 运行版本 A
experiment_a = evaluate(
    predict_v_a,
    data="qa_benchmark",
    evaluators=[exact_match],
    experiment_prefix="Version A",
    description="基准版本"
)

# 运行版本 B
experiment_b = evaluate(
    predict_v_b,
    data="qa_benchmark",
    evaluators=[exact_match],
    experiment_prefix="Version B",
    description="改进的提示词"
)

# 对比结果
print("=== 版本对比 ===")
print(f"版本 A 准确率: {calculate_accuracy(experiment_a):.2%}")
print(f"版本 B 准确率: {calculate_accuracy(experiment_b):.2%}")

def calculate_accuracy(experiment):
    scores = [r.evaluation_results[0].score for r in experiment.results]
    return sum(scores) / len(scores)

from scipy import stats

# 获取两个版本的评分
scores_a = [r.evaluation_results[0].score for r in experiment_a.results]
scores_b = [r.evaluation_results[0].score for r in experiment_b.results]

# T 检验
t_stat, p_value = stats.ttest_ind(scores_a, scores_b)

print(f"T 统计量: {t_stat:.4f}")
print(f"P 值: {p_value:.4f}")

if p_value < 0.05:
    if sum(scores_b) > sum(scores_a):
        print("✅ 版本 B 显著更好（p < 0.05）")
    else:
        print("✅ 版本 A 显著更好（p < 0.05）")
else:
    print("⚠️  差异不显著（p >= 0.05）")

from langsmith import Client
from datetime import datetime, timedelta

client = Client()

# 监控指标收集
def monitor_agent_health():
    """Agent 健康监测"""

    # 获取最近 24 小时的运行
    yesterday = datetime.now() - timedelta(days=1)
    runs = client.list_runs(
        project_name="my_agent",
        start_time=yesterday
    )

    # 计算关键指标
    total_runs = len(runs)
    successful_runs = sum(1 for run in runs if run.status == "success")
    error_runs = sum(1 for run in runs if run.status == "error")

    avg_duration = sum(
        (run.end_time - run.start_time).total_seconds()
        for run in runs
    ) / total_runs

    total_cost = sum(run.cost for run in runs if run.cost)

    # 计算错误率
    error_rate = error_runs / total_runs if total_runs > 0 else 0

    # 计算成本效率
    cost_per_success = total_cost / successful_runs if successful_runs > 0 else 0

    # 返回指标
    return {
        "total_runs": total_runs,
        "success_rate": successful_runs / total_runs,
        "error_rate": error_rate,
        "avg_duration": avg_duration,
        "total_cost": total_cost,
        "cost_per_success": cost_per_success
    }

metrics = monitor_agent_health()

print("=== Agent 健康监测 ===")
print(f"总请求: {metrics['total_runs']}")
print(f"成功率: {metrics['success_rate']:.2%}")
print(f"错误率: {metrics['error_rate']:.2%}")
print(f"平均耗时: {metrics['avg_duration']:.2f}s")
print(f"总成本: ${metrics['total_cost']:.2f}")
print(f"成本/成功: ${metrics['cost_per_success']:.4f}")

def check_health_alerts():
    """检查是否有告警条件"""
    metrics = monitor_agent_health()

    alerts = []

    # 错误率过高
    if metrics["error_rate"] > 0.1:  # > 10%
        alerts.append(f"⚠️  错误率过高: {metrics['error_rate']:.2%}")

    # 响应时间过长
    if metrics["avg_duration"] > 30:  # > 30秒
        alerts.append(f"⚠️  平均响应时间过长: {metrics['avg_duration']:.2f}s")

    # 成本突增
    if metrics["cost_per_success"] > 0.01:  # > $0.01 per success
        alerts.append(f"⚠️  成本过高: ${metrics['cost_per_success']:.4f}")

    if alerts:
        print("=== 告警 ===")
        for alert in alerts:
            print(alert)
    else:
        print("✅ 系统健康")

# 查找用户标记为"不好"的运行
bad_runs = client.list_runs(
    project_name="my_agent",
    filter='feedback_key:"user_rating" AND feedback_score:0'
)

print(f"找到 {len(bad_runs)} 条差评运行")

# 分析失败原因
failure_patterns = {}

for run in bad_runs:
    # 获取失败的工具调用
    if run.child_runs:
        for child in run.child_runs:
            if child.status == "error":
                tool_name = child.name
                if tool_name not in failure_patterns:
                    failure_patterns[tool_name] = 0
                failure_patterns[tool_name] += 1

# 排序并显示
for tool, count in sorted(failure_patterns.items(), key=lambda x: x[1], reverse=True):
    print(f"{tool}: {count} 次失败")

1. 基准测试 → 获得当前性能基线
   ↓
2. 假设提出 → 例如"改进提示词可以提高准确率 5%"
   ↓
3. 制定对策 → 编写新的提示词或改进模型选择
   ↓
4. 运行 A/B 测试 → 对比新旧版本
   ↓
5. 分析结果 → 是否满足假设？
   ├─ 是 → 更新版本到生产
   └─ 否 → 返回步骤 2，尝试新假设
   ↓
6. 监控上线 → 持续监控生产指标
   ↓
7. 收集反馈 → 用户标注、问题报告
   ↓
8. 返回步骤 1 → 开始新一轮优化

from langsmith import Client
from langsmith.evaluation import evaluate
from datetime import datetime

class ContinuousOptimization:
    def __init__(self, project_name: str):
        self.client = Client()
        self.project_name = project_name
        self.iteration = 0

    def baseline(self):
        """获得基准测试结果"""
        results = evaluate(
            predict,
            data="qa_benchmark",
            evaluators=[exact_match, relevance],
            experiment_prefix=f"Baseline_{datetime.now().isoformat()}",
            description="基准版本"
        )

        accuracy = self.calculate_accuracy(results)
        print(f"📊 基线准确率: {accuracy:.2%}")

        return results

    def propose_hypothesis(self, hypothesis: str):
        """提出假设"""
        self.hypothesis = hypothesis
        print(f"💡 假设: {hypothesis}")

    def implement_changes(self, changes: dict):
        """实现改进"""
        self.changes = changes
        print(f"⚙️  改进: {changes}")

    def run_experiment(self):
        """运行实验"""
        self.iteration += 1

        # 这里调用改进后的预测函数
        results = evaluate(
            predict_improved,  # 改进后的版本
            data="qa_benchmark",
            evaluators=[exact_match, relevance],
            experiment_prefix=f"Iteration_{self.iteration}_{datetime.now().isoformat()}",
            description=f"改进: {self.hypothesis}"
        )

        new_accuracy = self.calculate_accuracy(results)
        return new_accuracy

    def compare_with_baseline(self, baseline_acc, new_acc):
        """对比改进效果"""
        improvement = (new_acc - baseline_acc) / baseline_acc * 100

        if improvement > 0:
            print(f"✅ 改进成功! 提升 {improvement:.2%}")
            return True
        else:
            print(f"❌ 未达到预期，下降 {abs(improvement):.2%}")
            return False

    def deploy_to_production(self):
        """部署到生产环境"""
        print("🚀 部署新版本到生产")
        # 更新生产 Agent 的配置

    def calculate_accuracy(self, results):
        scores = [r.evaluation_results[0].score for r in results.results]
        return sum(scores) / len(scores)

# 使用示例
optimization = ContinuousOptimization("my_agent")

# 第 1 轮：基准
baseline_results = optimization.baseline()  # 准确率: 75%

# 第 2 轮：改进提示词
optimization.propose_hypothesis("详细的系统提示词能提高准确率")
optimization.implement_changes({"system_prompt": "新提示词..."})
acc_v2 = optimization.run_experiment()  # 准确率: 78%

if optimization.compare_with_baseline(0.75, acc_v2):
    optimization.deploy_to_production()

# 第 3 轮：切换模型
optimization.propose_hypothesis("使用更强的模型能进一步提高准确率")
optimization.implement_changes({"model": "gpt-4o"})
acc_v3 = optimization.run_experiment()  # 准确率: 82%

if optimization.compare_with_baseline(0.75, acc_v3):
    optimization.deploy_to_production()

┌─────────────────────────────────────────────────────────────┐
│                         用户接口层                             │
│                    (Web/API/Chat Interface)                   │
└─────────────┬───────────────────────────────────────────────┘
              │
┌─────────────▼───────────────────────────────────────────────┐
│                        应用服务层                              │
│   ┌─────────────┐  ┌──────────────┐  ┌─────────────┐       │
│   │ 查询处理器   │  │  Agent 引擎   │  │ 结果后处理   │       │
│   └─────────────┘  └──────────────┘  └─────────────┘       │
└─────────────┬───────────────────────────────────────────────┘
              │
┌─────────────▼───────────────────────────────────────────────┐
│                         核心层                                │
│   ┌────────────┐  ┌────────────┐  ┌─────────────┐          │
│   │ 向量检索器  │  │ 重排序器    │  │  LLM 网关    │          │
│   └────────────┘  └────────────┘  └─────────────┘          │
└─────────────┬───────────────────────────────────────────────┘
              │
┌─────────────▼───────────────────────────────────────────────┐
│                        数据层                                 │
│   ┌────────────┐  ┌────────────┐  ┌─────────────┐          │
│   │ 向量数据库  │  │ 文档存储    │  │  缓存层     │          │
│   └────────────┘  └────────────┘  └─────────────┘          │
└─────────────────────────────────────────────────────────────┘

from typing import List, Optional, Dict, Any
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Pinecone
from langchain_community.retrievers import ContextualCompressionRetriever
from langchain_community.retrievers.document_compressors import CohereRerank
from langchain_core.documents import Document
import redis
import hashlib
import json

class ProductionRAGSystem:
    """生产级 RAG 系统"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        self.vector_store = self._init_vector_store()
        self.cache = self._init_cache()
        self.llm = self._init_llm()
        self.retriever = self._init_retriever()

    def _init_vector_store(self):
        """初始化向量数据库"""
        return Pinecone.from_existing_index(
            index_name=self.config["pinecone_index"],
            embedding=self.embeddings,
            namespace=self.config.get("namespace", "default")
        )

    def _init_cache(self):
        """初始化缓存层"""
        return redis.Redis(
            host=self.config["redis_host"],
            port=self.config["redis_port"],
            db=0,
            decode_responses=True
        )

    def _init_llm(self):
        """初始化 LLM 网关"""
        return ChatOpenAI(
            model=self.config["llm_model"],
            temperature=0.7,
            max_retries=3,
            request_timeout=30
        )

    def _init_retriever(self):
        """初始化检索器（带重排序）"""
        base_retriever = self.vector_store.as_retriever(
            search_kwargs={"k": self.config.get("retrieve_k", 10)}
        )

        # 添加压缩/重排序（使用 CohereRerank）
        compressor = CohereRerank(
            model="rerank-multilingual-v3.0",
            top_n=self.config.get("top_k", 5)
        )

        return ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever
        )

    def _get_cache_key(self, query: str) -> str:
        """生成缓存键"""
        return f"rag:{hashlib.md5(query.encode()).hexdigest()}"

    def retrieve(self, query: str) -> List[Document]:
        """检索相关文档"""
        # 查询改写
        rewritten_query = self._rewrite_query(query)

        # 检索
        docs = self.retriever.get_relevant_documents(rewritten_query)

        # 后处理
        docs = self._post_process_docs(docs)

        return docs

    def _rewrite_query(self, query: str) -> str:
        """查询改写（提高检索质量）"""
        prompt = f"""请改写以下查询，使其更适合向量检索：

原始查询：{query}

改写后的查询（保持语义，优化关键词）："""

        response = self.llm.invoke(prompt)
        return response.content

    def _post_process_docs(self, docs: List[Document]) -> List[Document]:
        """文档后处理"""
        # 去重
        seen = set()
        unique_docs = []

        for doc in docs:
            doc_hash = hashlib.md5(doc.page_content.encode()).hexdigest()
            if doc_hash not in seen:
                seen.add(doc_hash)
                unique_docs.append(doc)

        # 按相关性重新排序
        unique_docs.sort(key=lambda x: x.metadata.get("score", 0), reverse=True)

        return unique_docs[:self.config.get("top_k", 5)]

    def generate_answer(self, query: str, docs: List[Document]) -> str:
        """生成答案"""
        context = "\n\n".join([doc.page_content for doc in docs])

        prompt = f"""基于以下上下文回答问题：

上下文：
{context}

问题：{query}

答案："""

        response = self.llm.invoke(prompt)
        return response.content

    def query(self, query: str, use_cache: bool = True) -> Dict[str, Any]:
        """完整查询流程"""
        # 检查缓存
        if use_cache:
            cache_key = self._get_cache_key(query)
            cached = self.cache.get(cache_key)

            if cached:
                return json.loads(cached)

        # 检索
        docs = self.retrieve(query)

        # 生成答案
        answer = self.generate_answer(query, docs)

        result = {
            "query": query,
            "answer": answer,
            "sources": [
                {
                    "content": doc.page_content[:200] + "...",
                    "metadata": doc.metadata
                }
                for doc in docs
            ]
        }

        # 写入缓存
        if use_cache:
            self.cache.setex(
                cache_key,
                self.config.get("cache_ttl", 3600),
                json.dumps(result)
            )

        return result