1. 项目概述：为什么我们需要一个Agent的“行为监控官”？

如果你正在构建基于大语言模型的智能体应用，无论是客服机器人、数据分析助手还是自动化工作流，那么下面这个场景你一定不陌生：某个版本更新后，Agent在测试环境表现良好，但一上线，回答质量就莫名其妙地下降，或者在某些边缘场景下开始“胡言乱语”。更头疼的是，你往往只能从用户的投诉中后知后觉，却很难在问题发生时立刻定位——是提示词的问题？是工具调用的逻辑错误？还是模型本身“抽风”了？

传统的应用监控（APM）工具盯着的是服务器的CPU、内存和请求延迟，但它们看不懂Agent内部“思考”的轨迹。而Judgeval，这个开源的Python SDK，就是为了填补这个空白而生的。你可以把它理解为专为AI Agent设计的“行为监控官”和“质检员”。它不关心你的服务器是否健康，它只关心你的Agent是否在“正确地思考和工作”。

它的核心价值在于，将Agent的每一次内部决策——从接收用户输入、调用工具、生成LLM请求到最终输出——都变成可观测、可追溯、可评估的“痕迹”。通过集成OpenTelemetry标准，Judgeval能无侵入地捕获这些痕迹，并允许你基于业务逻辑定义各种“评分标准”（例如答案相关性、事实一致性、指令遵循度），对线上流量进行实时异步评分，甚至设置警报。简单说，它让Agent的“黑盒”过程变得透明且可度量。

2. 核心架构与设计思路拆解

Judgeval的设计非常清晰，它不是一个重型的全链路平台，而是一个轻量级、可插拔的SDK。它的架构围绕几个核心概念展开，理解这些概念是高效使用它的关键。

2.1 基于OpenTelemetry的追踪体系

这是Judgeval的基石。OpenTelemetry是云原生领域事实上的可观测性标准。Judgeval没有另起炉灶，而是选择在其之上构建，这带来了巨大的兼容性优势。你的Agent产生的追踪数据，可以无缝对接你现有的Jaeger、Zipkin、Datadog或New Relic等观测平台。

关键设计考量 ：

  • 无侵入性 ：通过 @Tracer.observe() 装饰器或 wrap() 函数包装LLM客户端，业务代码几乎零改动。
  • 上下文传播 ：自动处理跨函数、跨线程的调用链，将一个用户会话中的所有步骤关联到一个“轨迹”中。
  • 丰富元数据 ：不仅记录输入输出，还能自动捕获LLM调用的令牌使用量、成本估算（如果提供商支持），这是进行成本分析和优化的重要依据。

2.2 双模评估：在线监控与离线评估

这是Judgeval最实用的功能分层，对应着不同的使用场景。

  • 在线监控 ：核心方法是 Tracer.async_evaluate() 。它的设计非常巧妙——评估逻辑在服务端异步执行。这意味着评估不会阻塞你Agent的实时响应，对终端用户完全无感。你可以用它来对生产环境的每一次交互进行质量打分，比如检查回答是否包含敏感信息、是否符合品牌语调等。一旦分数低于阈值，可以立即触发Slack或邮件告警。
  • 离线评估 ：通过 Judgeval 客户端的 evaluation.run() 方法进行。这通常用于批处理场景，例如：
    • 版本发布前 ：用一批“黄金标准”测试用例，对比新老版本Prompt或模型的效果。
    • 模型选型 ：用同一批数据测试GPT-4、Claude-3、本地模型的表现。
    • 数据集构建 ：人工标注一批数据后，用离线评估来验证自动评估器的准确性。

2.3 可扩展的“法官”系统

Judgeval内置了一些通用的评估器，如 faithfulness （忠实性）、 answer_relevancy （答案相关性）。但真正强大的地方在于，它允许你定义自己的“法官”。

“法官”的本质 ：一个继承自 Judge 基类的Python类，其 score 方法接收一个 Example 数据对象，并返回一个结构化的评分结果。Judgeval支持三种结果类型，覆盖了绝大多数评估场景：

  1. BinaryResponse ：是非判断。例如“回答是否包含特定关键词”、“工具调用是否成功”。
  2. NumericResponse ：连续分数。例如“回答质量打分（0-1）”、“与标准答案的相似度”。
  3. CategoricalResponse ：分类标签。例如“将用户意图分类为‘咨询’、‘投诉’或‘售前’”。

安全执行环境 ：当你通过CLI上传自定义法官代码时，Judgeval的后端会在一个安全的Firecracker微虚拟机中运行它。这隔离了你的代码与Judgeval的核心服务，确保了系统的安全性和稳定性，即使你的评估逻辑有Bug也不会拖垮整个服务。

3. 从零开始的实战部署与集成

理论讲完了，我们动手把它用起来。假设我们正在构建一个“智能产品咨询助手”。

3.1 环境准备与初始化

首先，安装SDK并配置凭证。建议使用虚拟环境。

# 创建并激活虚拟环境（可选但推荐）
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate  # Windows

# 安装judgeval
pip install judgeval

# 设置环境变量
export JUDGMENT_API_KEY="your_api_key_here"
export JUDGMENT_ORG_ID="your_org_id_here"

注意 ： JUDGMENT_API_KEY 和 JUDGMENT_ORG_ID 需要去 Judgment平台 注册免费账户获取。不要在代码中硬编码密钥，务必使用环境变量或密钥管理服务。

接下来，在应用启动时初始化Tracer。这通常在FastAPI/Django的启动文件或主脚本开头进行。

# app/core/monitoring.py
from judgeval import Tracer

def setup_monitoring():
    Tracer.init(
        project_name="product-consulting-agent", # 项目名，用于在平台分组
        endpoint="https://api.judgmentlabs.ai",  # 默认即可
    )
    print("Judgeval tracer initialized.")

3.2 基础链路追踪与LLM集成

现在，让我们改造一个简单的Agent函数。假设我们有一个函数，它先搜索产品知识库，再让LLM生成回答。

改造前 ：

from openai import OpenAI
import my_vector_db_module

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def answer_question(question: str) -> str:
    # 1. 搜索知识库
    context = my_vector_db_module.search(question, top_k=3)
    # 2. 请求LLM
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一个专业的产品顾问，请根据提供的知识库内容回答问题。"},
            {"role": "user", "content": f"知识库：{context}\n\n问题：{question}"}
        ]
    )
    return response.choices[0].message.content

改造后 ：

from judgeval import Tracer, wrap
from openai import OpenAI
import my_vector_db_module
import os

# 1. 包装LLM客户端，使其调用自动被追踪
client = wrap(OpenAI(api_key=os.getenv("OPENAI_API_KEY")))

# 2. 用装饰器标记工具函数
@Tracer.observe(span_type="tool")  # span_type可自定义，用于分类
def search_knowledge_base(question: str) -> str:
    """被追踪的搜索工具"""
    results = my_vector_db_module.search(question, top_k=3)
    # 这里可以记录更多自定义属性
    Tracer.set_attributes({"search.top_k": 3, "search.results_count": len(results)})
    return "\n".join(results)

# 3. 用装饰器标记主要的Agent函数
@Tracer.observe(span_type="agent")
def answer_question(question: str) -> str:
    """被追踪的Agent主函数"""
    context = search_knowledge_base(question)
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一个专业的产品顾问，请根据提供的知识库内容回答问题。"},
            {"role": "user", "content": f"知识库：{context}\n\n问题：{question}"}
        ]
    )
    answer = response.choices[0].message.content
    
    # 4. （可选）立即触发一次异步评估
    Tracer.async_evaluate(
        "answer_relevancy", # 使用平台预置的“答案相关性”评估器
        {"input": question, "actual_output": answer, "context": context}
    )
    
    return answer

完成以上步骤后，运行你的Agent。每次调用 answer_question ，一条完整的追踪轨迹（包含搜索工具调用和LLM调用）就会被发送到Judgment平台。你可以在仪表盘上清晰地看到函数耗时、令牌用量以及调用链关系。

3.3 集成到现有框架

如果你在使用更高级的Agent框架，Judgeval提供了开箱即用的集成。

LangGraph集成 ： 如果你用LangGraph构建有状态的Agent工作流，集成非常简单。

from langgraph.graph import StateGraph, END
from judgeval.integrations import Langgraph

# 在你的LangGraph应用初始化时调用
Langgraph.initialize()  # 这行代码会自动为你的Graph中的所有节点添加追踪

# 之后，你正常构建和运行你的Graph即可，所有节点执行都会被自动捕获。

其他框架 ： 对于Claude Agent SDK或使用了OpenLit的项目，类似的，只需导入对应的集成模块并调用 initialize 或 setup 方法。这种设计使得将现有项目迁移到Judgeval的监控体系下，成本极低。

4. 构建自定义评估逻辑与上线监控

内置评估器虽好，但真正的业务需求往往是独特的。我们来创建一个自定义的“法官”，检查Agent的回答是否在推荐具体产品时提到了产品的核心优势。

4.1 定义自定义法官

我们在项目中创建一个新文件 scorers/product_advantage_judge.py 。

# scorers/product_advantage_judge.py
from judgeval.judges import Judge
from judgeval.hosted.responses import BinaryResponse
from judgeval.data import Example
import re

class ProductAdvantageJudge(Judge[BinaryResponse]):
    """
    检查针对产品咨询的回答，如果提到了具体产品型号，
    则必须伴随提及该产品的至少一个核心优势。
    优势关键词列表可在初始化时配置。
    """
    def __init__(self, advantage_keywords: list[str]):
        super().__init__()
        self.advantage_keywords = advantage_keywords
        # 假设我们有一个产品型号的正则列表
        self.product_pattern = re.compile(r'\b(ProMax|Air|QuantumX)\b', re.IGNORECASE)

    async def score(self, data: Example) -> BinaryResponse:
        answer = data["actual_output"]
        input_question = data.get("input", "")
        
        # 1. 检查回答中是否提到了产品
        product_match = self.product_pattern.search(answer)
        if not product_match:
            # 如果没有提到产品，则本条规则不适用，直接返回通过（或可以定义为N/A）
            return BinaryResponse(value=True, reason="未提及特定产品，无需检查优势点。")
        
        mentioned_product = product_match.group(0)
        
        # 2. 检查是否包含任何优势关键词
        has_advantage = any(keyword.lower() in answer.lower() for keyword in self.advantage_keywords)
        
        if has_advantage:
            return BinaryResponse(
                value=True,
                reason=f"回答提到了产品'{mentioned_product}'，并包含了核心优势描述。"
            )
        else:
            return BinaryResponse(
                value=False,
                reason=f"回答提到了产品'{mentioned_product}'，但未发现核心优势关键词({self.advantage_keywords})。"
            )

# 注意：上传到Hosted Scorer时，Judgeval会实例化这个类。
# 我们需要告诉平台如何构造它。通常通过一个全局的`get_judge`函数。
def get_judge():
    # 这里可以从环境变量或配置中读取优势关键词
    keywords = ["长续航", "高清摄像头", "极速性能", "轻薄设计"]
    return ProductAdvantageJudge(advantage_keywords=keywords)

4.2 本地测试与上传

在将法官部署到云端前，强烈建议先在本地测试。

# test_judge_locally.py
from scorers.product_advantage_judge import ProductAdvantageJudge

async def test():
    judge = ProductAdvantageJudge(advantage_keywords=["长续航", "高清摄像头"])
    
    # 模拟一个Example对象
    from judgeval.data import Example
    test_data = Example.create(
        input="请推荐一款适合旅行的笔记本电脑。",
        actual_output="我推荐Air系列，它非常轻薄便携，适合旅行携带。",
        expected_output="" # 自定义法官可能不需要expected_output
    )
    
    result = await judge.score(test_data)
    print(f"评分结果: {result.value}")
    print(f"理由: {result.reason}")

import asyncio
asyncio.run(test())

本地测试通过后，使用CLI工具将其上传到Judgment平台，成为一个“托管评估器”。

# 初始化一个评分器模板（如果还没创建文件）
# judgeval scorer init -t binary -n ProductAdvantageJudge -o scorers/

# 上传自定义法官文件
judgeval scorer upload scorers/product_advantage_judge.py -p product-consulting-agent

# 上传成功后，平台会分配一个唯一的评分器ID或名称，记下它。

上传后，你可以在Judgment平台的“Scorers”部分看到它，并可以像使用内置评估器一样使用它。

4.3 在在线监控中调用自定义法官

现在，我们可以在生产环境的Agent中，异步调用这个自定义的质检逻辑。

@Tracer.observe(span_type="agent")
def answer_question(question: str) -> str:
    context = search_knowledge_base(question)
    response = client.chat.completions.create(...)
    answer = response.choices[0].message.content
    
    # 异步调用自定义的产品优势检查
    Tracer.async_evaluate(
        "product_advantage_judge",  # 你在平台上传后配置的评分器名称
        {
            "input": question,
            "actual_output": answer,
            # 可以传递额外的上下文，自定义法官的`score`方法可以通过data字典访问
            "context": context
        }
    )
    
    # 也可以同时调用多个评估器
    Tracer.async_evaluate(
        "answer_relevancy",
        {"input": question, "actual_output": answer}
    )
    
    return answer

4.4 配置警报规则

评估器上线后，你可以在Judgment平台上为它设置警报。例如，当 ProductAdvantageJudge 在连续10次调用中，失败率超过30%时，向团队的Slack频道发送一条告警信息。这能让你在用户大量投诉前，就意识到Agent的推荐话术可能出现了问题，需要检查最新的产品知识库或提示词。

5. 数据集管理与提示词版本化实战

监控和评估离不开高质量的数据。Judgeval提供了数据集和提示词版本化管理功能，这对于Agent的迭代优化至关重要。

5.1 构建与管理黄金测试集

“黄金测试集”是一组已知正确答案的高质量输入输出对，用于定期回归测试。

from judgeval import Judgeval
from judgeval.data import Example

client = Judgeval(project_name="product-consulting-agent")

# 创建或获取一个数据集
try:
    dataset = client.datasets.get(name="golden-set-v1")
except:
    # 如果不存在，则创建
    dataset = client.datasets.create(
        name="golden-set-v1",
        description="核心产品QA黄金测试集",
        examples=[
            Example.create(
                input="你们旗舰手机ProMax的电池容量是多少？",
                expected_output="ProMax的电池容量为5000mAh，支持100W有线快充。"
            ),
            Example.create(
                input="QuantumX游戏本有多重？",
                expected_output="QuantumX游戏本的重量约为2.3公斤。"
            ),
            Example.create(
                input="Air笔记本适合编程吗？",
                expected_output="Air笔记本搭载了高性能处理器和16GB内存，非常适合编程和日常开发工作。"
            ),
        ]
    )

# 在每次发布前，运行离线评估
evaluation = client.evaluation.create()
results = evaluation.run(
    examples=dataset.examples,  # 直接使用数据集中的例子
    scorers=["faithfulness", "answer_relevancy", "product_advantage_judge"],
    eval_run_name="pre-release-check-v2.1"
)

# 分析结果
for result in results:
    print(f"问题：{result.example.input}")
    for score in result.scores:
        print(f"  - {score.scorer_name}: {score.value} ({score.reason})")
    print("-" * 20)

5.2 实现提示词版本控制

提示词的微小改动可能对Agent行为产生巨大影响。Judgeval的提示词版本化功能，让你能像管理代码一样管理Prompt。

# 定义基础系统提示词模板
system_prompt_v1 = """
你是一个{{company}}的{{role}}。你的任务是{{task}}。
请始终保持{{tone}}的语气。如果用户询问产品信息，请参考以下知识：{{product_knowledge}}。
"""

# 将模板上传并标记为“测试”版本
prompt_v1 = client.prompts.create(
    name="system-prompt",
    prompt=system_prompt_v1,
    tags=["testing", "v1.0"],
    description="初始版本系统提示词"
)

# 当需要更新提示词时，创建新版本
system_prompt_v2 = system_prompt_v1 + "\n特别注意：在推荐产品时，必须提及其核心优势点。"
prompt_v2 = client.prompts.create(
    name="system-prompt",
    prompt=system_prompt_v2,
    tags=["production", "v1.1"],  # 更新标签
    description="优化版，强制要求提及产品优势"
)

# 在Agent代码中，动态获取指定版本的提示词并编译
def get_compiled_prompt(tag: str = "production"):
    prompt_obj = client.prompts.get(name="system-prompt", tag=tag)
    compiled = prompt_obj.compile(
        company="Acme Tech",
        role="高级产品顾问",
        task="准确、友好地回答客户关于我司产品的所有问题",
        tone="专业且热情",
        product_knowledge="ProMax: 长续航，高清摄像...\nAir: 轻薄，性能强..."
    )
    return compiled

# 在LLM调用中使用编译后的提示词
current_prompt = get_compiled_prompt("production") # 或 "testing"
messages = [{"role": "system", "content": current_prompt}, ...]

通过这种方式，你可以轻松地进行A/B测试。让10%的流量使用 v1.0 的提示词，90%使用 v1.1 ，然后在Judgment平台上对比两个版本在各项评估指标上的表现，用数据驱动决策。

6. 常见问题、性能考量与排查技巧

在实际使用中，你可能会遇到一些典型问题。以下是我在多个项目中落地Judgeval后总结的经验。

6.1 常见问题与解决方案

问题现象 可能原因 排查步骤与解决方案
仪表盘看不到追踪数据 1. 凭证未设置或错误。
2. Tracer.init() 未被调用。
3. 网络问题或SDK版本不兼容。		 1. 检查 JUDGMENT_API_KEY 和 JUDGMENT_ORG_ID 环境变量是否正确。
2. 确保应用启动时成功执行了 Tracer.init() 。
3. 运行 judgeval --version 检查版本，查看官方文档更新。尝试在代码中捕获并打印初始化错误。
async_evaluate 评估结果未显示 1. 评估器名称拼写错误。
2. 自定义评估器上传失败或未激活。
3. 评估过程本身抛出异常。		 1. 在Judgment平台“Scorers”页面核对评估器名称。
2. 检查自定义评估器上传日志，确保 get_judge 函数正确定义。
3. 在自定义评估器的 score 方法中添加详细的日志和异常捕获，先进行充分的本地测试。
追踪数据延迟高 1. SDK默认使用异步批量上报，有缓冲延迟。
2. 网络延迟。		 1. 这是正常设计，为了性能。通常延迟在几秒到一分钟。如需更实时，可查阅SDK配置项，调整批处理大小和间隔（如果支持）。
2. 检查服务器区域与Judgment服务的网络连通性。
集成LangGraph后部分节点未追踪 1. 集成代码未在Graph构建 前 执行。
2. 节点是异步函数，可能需要特殊处理。		 1. 确保 Langgraph.initialize() 在创建 StateGraph 之前调用。
2. 检查LangGraph官方文档和Judgeval集成文档，看是否有对异步节点的额外说明。
令牌数或成本计算不准 1. 使用的LLM提供商SDK版本较旧。
2. 模型不在Judgeval预置的计价列表中。		 1. 确保使用最新版的OpenAI、Anthropic等官方SDK，Judgeval依赖其返回的元数据。
2. 对于新模型或自定义模型，成本计算可能回退到估算。可在 Tracer.set_attributes() 中手动覆盖成本属性。

6.2 性能与成本最佳实践

  • 评估器设计原则 ：自定义评估器（尤其是使用LLM作为裁判的）可能是性能瓶颈和成本来源。尽量做到：
    • 轻量 ：优先使用规则（正则、关键词）或小型分类模型。
    • 异步与批处理 ：离线评估时，利用 Judgeval 客户端的批量执行能力。在线监控的 async_evaluate 本身是异步的，不用担心。
    • 缓存 ：对于相同或相似的输入，考虑在评估器内部实现简单的缓存机制，避免重复计算。
  • 采样率控制 ：在生产环境，对100%的流量进行所有评估可能不经济。可以在 Tracer.async_evaluate() 调用前添加采样逻辑。 
    import random

def should_sample(samplerate: float = 0.2) -> bool:
    return random.random() < samplerate

if should_sample(0.1): # 只对10%的请求进行资源密集型评估
    Tracer.async_evaluate("expensive_llm_judge", data)
  • 关注数据体积 ：虽然单条追踪数据不大，但高并发下总量可观。定期检查仪表盘，关注数据摄入量。对于调试完成的、稳定的工具函数，可以考虑移除 @Tracer.observe 装饰器，或只追踪更高层级的Agent函数。

6.3 调试技巧

  1. 本地开发模式 ：在开发时，可以设置环境变量 JUDGMENT_ENV=local 或查看SDK是否支持将追踪日志输出到控制台，而不是发送到远程服务器，方便调试。
  2. 善用属性 ：在追踪的Span中，使用 Tracer.set_attributes() 记录关键的中间变量或业务状态。当在仪表盘查看轨迹详情时，这些属性是排查问题的金钥匙。
  3. 关联业务ID ：在追踪开始时，设置一个唯一的业务ID（如用户会话ID、订单号），这样你可以在日志系统和Judgment仪表盘中通过同一个ID串联所有信息，实现端到端的排查。 
    @Tracer.observe(span_type="agent")
def handle_user_session(session_id: str, question: str):
    Tracer.set_attributes({"business.session_id": session_id})
    # ... 后续处理

将Judgeval融入你的Agent开发流程，初期可能会觉得多了一些步骤，但一旦建立起“追踪-评估-警报”的闭环，你就会发现它对Agent的稳定性、可解释性和迭代速度带来的提升是巨大的。它让你从被动救火转向主动运维，真正地“看见”你的Agent是如何工作的。

Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

cover

Agent沙箱实战：Shell工具调用的路径白名单与零信任逃逸审计

avatar 龙虾开发者社区
cover

HiClaw网关的审批流设计：如何避免工具调用中的误操作灾难？

avatar 龙虾开发者社区
cover

Agent 网关密钥管理：为什么你的硬件绑定总在 session 失效时翻车

avatar 龙虾开发者社区