如何创建复合评估器

import os
from dotenv import load_dotenv
from openai import OpenAI
from langsmith import Client
from pydantic import BaseModel
import json

# 从 .env 文件加载环境变量
load_dotenv()

# 访问环境变量
openai_api_key = os.getenv('OPENAI_API_KEY')
langsmith_api_key = os.getenv('LANGSMITH_API_KEY')
langsmith_project = os.getenv('LANGSMITH_PROJECT', 'default')


# 创建数据集。只需执行一次。
client = Client()
oai_client = OpenAI()

examples = [
  {
    "inputs": {"blog_intro": "今天，我们很高兴地宣布 LangSmith 正式可用——这是我们专为部署和扩展长期运行、有状态代理而打造的基础设施和管理层。自去年 6 月进入测试阶段以来，已有近 400 家公司使用 LangSmith 将其代理部署到生产环境中。代理部署是交付可靠代理时面临的下一个重大难题，而 LangSmith 通过以下功能显著降低了这一门槛：一键部署，几分钟内即可上线；30 个 API 端点，用于设计适应各种交互模式的自定义用户体验；水平扩展，以处理突发的长期运行流量；持久化层，为记忆、对话历史以及人机协作或多代理工作流提供支持；原生 Studio（代理 IDE），便于调试、查看和迭代。"},
  },
  {
    "inputs": {"blog_intro": "Klarna 以消费者为中心、由 AI 驱动的支付和购物解决方案重塑了全球商业。Klarna 平台拥有超过 8500 万活跃用户，每日处理 250 万笔交易，是一家金融科技领域的领导者，在简化购物流程的同时，为消费者提供更智能、更灵活的金融解决方案。Klarna 的旗舰产品 AI Assistant 正在彻底改变购物和支付体验。这个 AI Assistant 基于 LangGraph 构建，由 LangSmith 提供支持，可处理从客户付款、退款到其他支付升级等各种任务。截至目前，AI Assistant 已处理 250 万次对话。它不仅仅是一个聊天机器人，更是一个变革性的代理，能够完成相当于 700 名全职员工的工作，快速交付成果并提升公司的效率。"},
  },
]

dataset = client.create_dataset(dataset_name="Blog Intros")

client.create_examples(
  dataset_id=dataset.id,
  examples=examples,
)

# 定义目标函数。在此示例中，我们使用一个简单的函数，根据博客简介生成一条推文。
def generate_tweet(inputs: dict) -> dict:
    instructions = (
      "请根据博客简介，生成一条吸引人且专业的推文，用于在社交媒体上推广这篇博客文章。在推文中总结博客文章的核心观点。请恰当地使用表情符号。"
    )
    messages = [
        {"role": "system", "content": instructions},
        {"role": "user", "content": inputs["blog_intro"]},
    ]
    result = oai_client.responses.create(
        input=messages, model="gpt-5-nano"
    )
    return {"tweet": result.output_text}

# 定义评估器。在此示例中，我们使用三个评估器：摘要、格式和语气。
def summary(inputs: dict, outputs: dict) -> bool:
    """判断这条推文是否很好地概括了博客简介。"""
    instructions = "请根据以下文本和摘要，判断该摘要是否很好地概括了原文。"

    class Response(BaseModel):
        summary: bool

    msg = f"问题：{inputs['blog_intro']}\n答案：{outputs['tweet']}"
    response = oai_client.responses.parse(
        model="gpt-5-nano",
        input=[{"role": "system", "content": instructions,}, {"role": "user", "content": msg}],
        text_format=Response
    )

    parsed_response = json.loads(response.output_text)
    return parsed_response["summary"]

def formatting(inputs: dict, outputs: dict) -> bool:
    """判断这条推文的格式是否便于人类阅读。"""
    instructions = "请根据以下文本，判断其格式是否良好，确保人类可以轻松阅读。请特别注意间距和标点符号。"

    class Response(BaseModel):
        formatting: bool

    msg = f"{outputs['tweet']}"
    response = oai_client.responses.parse(
        model="gpt-5-nano",
        input=[{"role": "system", "content": instructions,}, {"role": "user", "content": msg}],
        text_format=Response
    )

    parsed_response = json.loads(response.output_text)
    return parsed_response["formatting"]

def tone(inputs: dict, outputs: dict) -> bool:
    """判断这条推文的语气是否具有信息性、友好且引人入胜。"""
    instructions = "请根据以下文本，判断这条推文是否具有信息性，同时语气友好且引人入胜。"

    class Response(BaseModel):
        tone: bool

    msg = f"{outputs['tweet']}"
    response = oai_client.responses.parse(
        model="gpt-5-nano",
        input=[{"role": "system", "content": instructions,}, {"role": "user", "content": msg}],
        text_format=Response
    )
    parsed_response = json.loads(response.output_text)
    return parsed_response["tone"]

# 使用数据集、目标函数和评估器调用 evaluate()。
results = client.evaluate(
    generate_tweet,
    data=dataset.name,
    evaluators=[summary, tone, formatting],
    experiment_prefix="gpt-5-nano",
)

# 获取实验名称，下一节将在 client.get_experiment_results() 中使用
experiment_name = results.experiment_name

from typing import Dict
import math
from langsmith import Client
from dotenv import load_dotenv

load_dotenv()

# TODO: 替换为你的实验名称。可在 UI 中找到，或从上面的 client.evaluate() 结果中获取
YOUR_EXPERIMENT_NAME = "placeholder_experiment_name"

# 设置各个评估器分数的权重
DEFAULT_WEIGHTS: Dict[str, float] = {
    "summary": 0.7,
    "tone": 0.2,
    "formatting": 0.1,
}
WEIGHTED_FEEDBACK_NAME = "weighted_summary"

# 获取实验结果
client = Client()
results = client.get_experiment_results(
    name=YOUR_EXPERIMENT_NAME,
)

# 计算每个运行的加权分数
def calculate_weighted_score(feedback_stats: dict) -> float:
    if not feedback_stats:
        return float("nan")

    # 检查所有必需的指标是否存在且包含数据
    required_metrics = set(DEFAULT_WEIGHTS.keys())
    available_metrics = set(feedback_stats.keys())

    if not required_metrics.issubset(available_metrics):
        return float("nan")

    # 计算加权分数
    total_score = 0.0
    for metric, weight in DEFAULT_WEIGHTS.items():
        metric_data = feedback_stats[metric]
        if metric_data.get("n", 0) > 0 and "avg" in metric_data:
            total_score += metric_data["avg"] * weight
        else:
            return float("nan")

    return total_score

# 处理每个运行并写入反馈
# 请注意，实验结果需要完成处理后才能调用此代码。
for example_with_runs in results["examples_with_runs"]:
    for run in example_with_runs.runs:
        if run.feedback_stats:
            score = calculate_weighted_score(run.feedback_stats)
            if not math.isnan(score):
                client.create_feedback(
                    run_id=run.id,
                    key=WEIGHTED_FEEDBACK_NAME,
                    score=float(score),
                    session_id=run.session_id,
                )