LiveKit Agents 如何用 JudgeGroup 在会话结束后对对话质量打分
【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents
你写了一个语音 Agent 并已经能跑通对话,但每次挂断电话后,这通会话到底表现得好不好,只能靠人回听录音判断。LiveKit Agents 的livekit.agents.evals模块提供JudgeGroup:在会话结束回调on_session_end里,把整段聊天记录交给一组评审 Judge,让 LLM 按各自标准逐项打分(pass/fail/maybe),并把结果自动打到会话标签上。仓库里的 frontdesk 示例 就是这套用法的完整落地,下文按其真实代码拆解操作路径。
机制:JudgeGroup 在会话结束时做了什么
核心实现在 evaluation.py 与 judge.py:
JudgeGroup.evaluate(chat_ctx)会并发运行组内所有 Judge,每个 Judge 返回一个JudgmentResult,包含verdict(pass/fail/maybe)、reasoning和所用标准instructions。- 单个 Judge 抛异常时只记录 warning 并从结果中剔除,不会打断整组评估。
- 返回的
EvaluationResult提供聚合判断:score(0.0~1.0,pass=1、maybe=0.5、fail=0)、all_passed、any_passed、majority_passed、none_failed。其中maybe在all_passed中按未通过处理。 - 关键行为:如果
evaluate()在 job 上下文中运行,结果会自动通过ctx.tagger._evaluation(...)写入会话标签,格式为lk.judge.<judge名>:<verdict>,并在会话结束时随其他标签一起上传到 LiveKit Cloud(见 observability.py 中 Tagger 的实现)。不在 job 上下文中调用时跳过打标,但不影响返回结果。
每条判定内部通过 LLM 的 function calling 完成:Judge 必须调用submit_verdict(verdict, reasoning)工具返回结论(tool_choice="required"),模型连接超时为 90 秒;除模型名含gpt-5外都会设置temperature=0.0。
准备条件
- 一个已能通过
@server.rtc_session注册入口的 LiveKit Agents 应用(frontdesk 示例的结构见 agent.py)。 - 给
JudgeGroup的llm传模型字符串(如"openai/gpt-4o-mini")时走 LiveKit 推理网关,需要配置LIVEKIT_API_KEY和LIVEKIT_API_SECRET环境变量(tests/test_judge.py 中即如此准备环境)。传一个自行构造的 LLM 实例则不需要。 - 模型字符串会自动取该模型支持的最低 reasoning effort,并固定为
low推理优先级,避免打分流量与在线语音会话争抢网关配额;想覆盖这些设置就传配置好的 LLM 实例而不是字符串。
操作步骤:在 on_session_end 中接入 JudgeGroup
以 frontdesk/agent.py 的真实代码为主线。
1. 在会话结束回调中拿到聊天记录
入口函数通过@server.rtc_session(on_session_end=on_session_end)注册。回调里第一步是生成会话报告并取出chat_history:
async def on_session_end(ctx: JobContext) -> None: # `on_session_end` runs even if the job crashed before the AgentSession # started (e.g. a bad timezone, a calendar fault) — make_session_report # raises in that case, and there's nothing to evaluate anyway. try: report = ctx.make_session_report() except RuntimeError: return注意这里的边界:即使 job 在AgentSession启动前就崩溃,on_session_end仍会执行,此时make_session_report()会抛RuntimeError,直接return即可——没有会话内容也就没有可评估的对象。报告类型SessionReport定义在 report.py,chat_history字段就是ChatContext,还包含 job/room 信息、会话选项快照和模型用量。
frontdesk 还加了一条长度护栏,避免对极短的会话浪费一次 LLM 调用:
chat = report.chat_history.copy(exclude_function_call=True, exclude_instructions=True) if len(chat.items) < 3: return2. 构造 JudgeGroup 并选择评审项
frontdesk 使用了全部 8 个内置 Judge:
judges = JudgeGroup( llm="openai/gpt-4o-mini", judges=[ task_completion_judge(), accuracy_judge(), tool_use_judge(), handoff_judge(), safety_judge(), relevancy_judge(), coherence_judge(), conciseness_judge(), ], ) await judges.evaluate(report.chat_history)导入方式为from livekit.agents.evals import JudgeGroup, accuracy_judge, ...。各内置 Judge 的判定标准(摘自 judge.py 的 docstring):
| Judge | 名称 | 判定标准 |
|---|---|---|
task_completion_judge | task_completion | Agent 是否完成其 instructions 中的目标;完成、妥善移交或合理拒绝算 pass,用户需求被忽略、无解决且未移交算 fail。适用于客服、预约、订单管理 |
accuracy_judge | accuracy | 信息必须准确且有依据;陈述与工具输出不符、捏造细节、误报姓名/日期/数字等算 fail。适用于医疗、保险、金融 |
tool_use_judge | tool_use | 工具选型、参数、输出解读与错误处理是否正确;本就不需要工具时算 pass |
handoff_judge | handoff | 跨 Agent 移交是否保留上下文;会话中没有发生移交时直接自动 pass |
safety_judge | safety | 是否越权给医疗/法律/财务建议、违规泄露信息、该升级人工时未升级、语言有害 |
relevancy_judge | relevancy | 回应是否切题,是否忽视用户输入或跑题 |
coherence_judge | coherence | 表达是否连贯有逻辑、不自相矛盾 |
conciseness_judge | conciseness | 是否简洁,有无冗余重复——对语音场景尤其重要 |
不需要全上:judges列表传哪几个由你的成功标准决定,frontdesk 是全部启用的特例。
3. 验证打分结果
有三种文档支持的查看方式:
打印到控制台:设置环境变量
LIVEKIT_EVALS_VERBOSE=1(evaluation.py中读取该变量,默认 0),evaluate()完成后会打印每个 Judge 的 verdict 和 reasoning,格式如下(文档实际输出格式,内容为示例性质):+ JudgeGroup evaluation results: [task_completion] verdict=pass reasoning: ...读会话标签:评估完成后,每个 Judge 对应一个
lk.judge.<name>:<verdict>标签。frontdesk 在回调末尾打印并依据业务状态补充业务标签:userdata = ctx.primary_session.userdata if userdata.cal.scheduled_appointments: ctx.tagger.success() else: ctx.tagger.fail(reason="Appointment was not booked") logger.info("session tags: %s", ctx.tagger.tags)ctx.tagger.evaluations属性还能拿到结构化的评估记录列表,每项含name、tag、verdict、reasoning、instructions。程序化判断:
evaluate()的返回值EvaluationResult可直接在代码里用result.score、result.all_passed等属性做后续动作(如触发告警)。
可选分支
对照参考对话打分:
evaluate(chat_ctx, reference=...)接受一个参考ChatContext,各 Judge 会把 reference 一并放进提示词作为对照。frontdesk 未使用,适用于你有一份标准示范对话的场景。写自定义 Judge:不依赖 LLM 的确定性检查继承
Judge并覆写evaluate,judge.py 中的示例是检查回复是否包含引用标记:class CitationJudge(Judge): def __init__(self): super().__init__(name="citation") async def evaluate(self, *, chat_ctx, reference=None, llm=None): has_citation = any( "[source]" in (m.text_content or "") for m in chat_ctx.messages ) return JudgmentResult( verdict="pass" if has_citation else "fail", reasoning="Found citation markers" if has_citation else "No citations", )任何实现了
name属性和evaluate()的对象都满足Evaluator协议,可以直接放进judges列表与内置 Judge 混用。
边界与限制
maybe不算通过:JudgmentResult.passed仅在 verdict 为pass时为真,all_passed会把maybe视为未通过,读分数时要按 0.5 计。- 某个 Judge 调用失败只记 warning 并被排除出结果,
EvaluationResult里只会剩成功的判定,不会报错中断。 - 打标只在 job 上下文内发生;脱离 job 环境(比如单测里直接调用)时
evaluate()仍返回完整结果,只是没有lk.judge.*标签。 handoff_judge在会话没有实际移交时短路返回 pass,不要把它当作“移交体验良好”的证据,只能说明该会话无需检查移交。- 模型字符串路径的最低 reasoning effort 与
low推理优先级是刻意设计(见 evaluation.py 的构造函数),打分流量不应与在线语音争抢网关配额;需要更高推理强度时请传 LLM 实例。
完整可运行参考:examples/frontdesk/agent.py 的on_session_end(L290–L326),以及 examples/frontdesk/README.md 中 “Evaluations” 一节的说明。想核对内部行为(工具强制调用、temperature 设置、reasoning effort 映射)时可参考 tests/test_judge.py。
【免费下载链接】agentsA framework for building realtime voice AI agents 🤖🎙️📹项目地址: https://gitcode.com/GitHub_Trending/agen/agents
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考