什么是 ‘Self-Evaluation Metrics’:让 Agent 节点自己输出执行质量评分,用于后续的自动化筛选

各位同仁,各位对人工智能充满热情的开发者与研究者们,大家下午好!

今天,我们将深入探讨一个在构建智能体(Agent)系统时至关重要,却又常常被低估的领域:自评估指标(Self-Evaluation Metrics)。具体来说,我们将聚焦于如何让智能体节点本身输出其执行质量评分,并探讨这些评分如何为后续的自动化筛选、优化乃至自主学习提供坚实基础。

在当今快速发展的AI时代,我们正从构建单一功能的模型转向构建能够感知、决策、行动并学习的复杂智能体系统。这些系统通常由多个互联的智能体节点组成,每个节点可能负责一个特定的子任务。然而,当这些智能体独立运行时,我们如何有效地监控它们的表现?如何快速识别并淘汰低质量的执行结果?传统的人工评估或离线指标往往滞后且成本高昂。正是在这样的背景下,自评估指标的概念应运而生,它赋予了智能体自我批判、自我审视的能力。

一、 智能体自评估:为何需要它?

在探讨技术细节之前,我们首先需要理解为什么自评估如此重要。想象一下,你构建了一个复杂的智能体工作流,它可能涉及信息检索、内容生成、代码编写、数据分析等多个步骤。每个步骤都由一个或多个智能体模块完成。

传统评估方法的局限性:

  1. 滞后性与低效性: 传统的评估往往是离线的,需要人工介入或通过预定义的测试集进行。这意味着在发现问题时,智能体可能已经产生了大量的低质量输出,错过了实时纠正的最佳时机。
  2. 成本高昂: 特别是在生产环境中,如果每个任务执行都需要人工审核,那么随着任务量的增长,成本将呈几何级数上升。
  3. 颗粒度不足: 离线评估通常针对整个任务流程或某个阶段的最终结果。它很难精确地指出在智能体执行过程中的哪一步、哪个决策出了问题,从而难以进行有针对性的优化。
  4. 难以适应动态环境: 智能体系统往往部署在动态变化的环境中。预设的评估标准可能很快过时,而人工更新和调整这些标准既耗时又耗力。
  5. 缺乏对“不确定性”的感知: 智能体在执行任务时,可能面临各种不确定性,例如信息不完整、指令模糊等。传统评估无法让智能体主动表达其对自身结果的信心水平。

自评估的优势:

自评估机制旨在克服上述挑战,它让智能体在完成任务或任务的某个阶段后,能够对其自身的输出、决策过程乃至对任务的理解程度进行量化评估。这种内在的反馈循环带来了诸多益处:

  • 实时反馈与纠正: 智能体可以立即识别低质量或高风险的输出,并触发重试、修正或寻求人类帮助。
  • 自动化筛选与路由: 基于自评估分数,系统可以自动过滤掉不合格的结果,或者将高分结果直接投入生产,将中低分结果路由给人类专家进行复核。
  • 提升系统韧性: 智能体不再是盲目执行的机器,它能感知到自身的不足,从而表现出更高的容错性和健壮性。
  • 促进自主学习: 自评估分数可以作为强化学习中的奖励信号,指导智能体优化其决策策略。
  • 优化资源利用: 识别出无法完成或信心不足的任务,避免不必要的计算资源浪费。

简而言之,自评估是构建真正自主、可靠和可扩展智能体系统的基石。它将智能体从单纯的执行者转变为具备一定元认知能力的思考者。

二、 自评估的核心概念:执行质量评分

当提到“自评估指标”时,我们通常指的是智能体在完成一次特定任务执行后,为其结果或过程打出的“执行质量评分”。这个评分不是一个单一、通用的概念,它可以是多维度的,反映了智能体对自身表现的综合考量。

什么是“执行质量”?

执行质量涵盖了智能体输出的多个关键属性。这些属性可以根据具体的应用场景和任务目标进行定制。以下是一些常见的执行质量维度:

| 维度名称 | 描述 尽管这里强调的是“自评估”,但其背后的逻辑和策略与我们作为开发者评估他人工作或模型输出的逻辑是同源的。关键在于将这些评估逻辑“内化”到智能体自身,使其能够在执行过程中或执行结束后,根据预设的标准或学习到的模式进行自我评分。

三、 智能体自评估的架构模式

为了将自评估能力融入智能体系统,我们可以采用几种不同的架构模式。这些模式决定了评估发生在智能体执行周期的哪个阶段。

  1. 后执行评估器 (Post-Execution Scorer):

    • 描述: 这是最常见的模式。智能体首先完成其核心任务,生成最终输出。然后,一个独立的自评估模块(可以是智能体自身的一个子功能,也可以是另一个专门的“评估智能体”)接收这个输出和原始任务指令,对其进行分析并生成评分。
    • 优点: 结构清晰,与核心任务逻辑解耦;评估是在任务完成后进行的,可以对整个结果进行综合考量。
    • 缺点: 无法在任务执行过程中提供实时指导;如果任务执行时间很长,发现问题会比较晚。
    • 应用场景: 文案生成、代码编写、报告总结等,任务执行是一个相对独立的完整过程。
    # 架构示意 (Python 伪代码)
    class AgentTaskExecutor:
        def execute_task(self, prompt: str) -> str:
            print("Agent executing task...")
            # 模拟任务执行,生成输出
            output = f"Generated content for: '{prompt}'. This is a draft version."
            return output
    
    class SelfEvaluationModule:
        def evaluate_output(self, prompt: str, output: str) -> dict:
            print("Self-evaluation module assessing output...")
            # 模拟评估逻辑
            score = 0.0
            feedback = []
    
            # 简单规则:检查输出是否包含关键字 'draft'
            if "draft" in output.lower():
                score -= 0.2
                feedback.append("Output still seems to be a draft.")
    
            # 简单规则:检查输出长度
            if len(output) < 50:
                score -= 0.3
                feedback.append("Output is too short.")
    
            # 假设一个基础分数
            score += 0.7 
    
            # 进一步的LLM或工具辅助评估可以放在这里
            # ...
    
            return {"overall_score": max(0, score), "feedback": feedback, "confidence": max(0.5, 0.5 + score/2)}
    
    # 智能体工作流
    def agent_workflow_post_execution(prompt: str):
        executor = AgentTaskExecutor()
        scorer = SelfEvaluationModule()
    
        # 1. 任务执行
        task_output = executor.execute_task(prompt)
        print(f"Task Output: {task_output}")
    
        # 2. 后执行评估
        evaluation_results = scorer.evaluate_output(prompt, task_output)
        print(f"Self-Evaluation Results: {evaluation_results}")
    
        if evaluation_results["overall_score"] < 0.6:
            print("Output quality is low, considering retry or human review.")
        else:
            print("Output quality is acceptable, proceeding.")
    
        return task_output, evaluation_results
    
    # 示例调用
    # workflow_prompt = "Write a short marketing blurb for a new AI conference."
    # agent_workflow_post_execution(workflow_prompt)
  2. 执行中监控与评估器 (In-Execution Monitor/Scorer):

    • 描述: 在智能体执行任务的各个中间步骤或关键决策点,集成评估逻辑。智能体可以定期检查其当前状态、中间结果或下一步行动计划,并对其进行评分。如果发现偏差或低质量,可以立即调整策略、回溯或提前终止。
    • 优点: 提供实时反馈,允许智能体在任务完成前进行自我纠正,减少资源浪费,提高效率。
    • 缺点: 评估逻辑需要与任务执行过程紧密耦合,设计和实现更为复杂;可能引入额外的计算开销。
    • 应用场景: 多步骤规划、代码调试、复杂数据分析,需要迭代和试错的任务。
    class MultiStepAgent:
        def __init__(self):
            self.steps_completed = 0
            self.intermediate_results = []
    
        def _execute_step(self, step_instruction: str) -> str:
            self.steps_completed += 1
            print(f"  Executing step {self.steps_completed}: {step_instruction}")
            # 模拟步骤执行
            result = f"Result of step {self.steps_completed} for '{step_instruction}'."
            self.intermediate_results.append(result)
            return result
    
        def _self_monitor_step(self, step_instruction: str, step_output: str) -> dict:
            print(f"  Self-monitoring step {self.steps_completed} output...")
            score = 0.0
            feedback = []
    
            # 模拟评估:检查关键指令是否被考虑
            if "summarize" in step_instruction.lower() and "summary" not in step_output.lower():
                score -= 0.4
                feedback.append("Instruction to summarize might not be fully met.")
    
            score += 0.8 # 基础分
            return {"step_score": max(0, score), "step_feedback": feedback, "step_confidence": max(0.6, 0.6 + score/2)}
    
        def execute_multi_step_task(self, main_goal: str, steps: list[str]) -> tuple[str, dict]:
            overall_scores = []
            final_output = ""
            print(f"Agent starting multi-step task: {main_goal}")
    
            for i, step_instruction in enumerate(steps):
                step_output = self._execute_step(step_instruction)
                monitor_results = self._self_monitor_step(step_instruction, step_output)
                overall_scores.append(monitor_results)
    
                print(f"    Step {i+1} Monitor: {monitor_results}")
    
                if monitor_results["step_score"] < 0.5:
                    print(f"    Warning: Step {i+1} quality is low. Considering re-planning or retry.")
                    # 实际中可能触发重试、回溯或寻求人工帮助
                    # For simplicity, we just continue here.
                    pass
    
                final_output += f"n--- Step {i+1} Output ---n{step_output}"
    
            # 综合所有步骤的评分
            overall_task_score = sum(s["step_score"] for s in overall_scores) / len(overall_scores) if overall_scores else 0
            print(f"Multi-step task completed. Overall estimated quality: {overall_task_score:.2f}")
    
            return final_output, {"overall_task_score": overall_task_score, "step_details": overall_scores}
    
    # 示例调用
    # agent = MultiStepAgent()
    # goal = "Analyze a market trend and suggest a product idea."
    # plan_steps = [
    #     "Search for recent market reports on AI in healthcare.",
    #     "Identify key growth areas and challenges from the reports.",
    #     "Summarize potential product opportunities.",
    #     "Propose a concrete product idea based on the summary."
    # ]
    # final_result, evaluation = agent.execute_multi_step_task(goal, plan_steps)
    # print(f"nFinal Result:n{final_result}")
    # print(f"Overall Evaluation: {evaluation['overall_task_score']}")
  3. 预执行规划与预测 (Pre-Execution Planner/Scorer):

    • 描述: 在智能体开始执行任务之前,它会评估任务本身的难度、模糊性、所需资源,并预测其成功执行的可能性或预期的质量。这通常涉及对任务指令的理解、对自身能力的评估以及对外部工具可用性的考量。
    • 优点: 可以在任务开始前就识别出高风险或不可能完成的任务,避免资源浪费;有助于智能体选择最佳的执行策略或寻求帮助。
    • 缺点: 预测本身可能不准确;需要智能体具备较强的元认知能力和规划能力。
    • 应用场景: 复杂任务调度、资源受限环境、需要动态选择工具链的场景。
    class AgentPrePlanner:
        def __init__(self, available_tools: list[str]):
            self.available_tools = available_tools
    
        def predict_task_feasibility(self, prompt: str) -> dict:
            print(f"Agent analyzing prompt for pre-execution planning: '{prompt}'")
            difficulty_score = 0.5 # 默认中等难度
            confidence_score = 0.8 # 默认较高信心
            required_tools = []
    
            # 模拟基于关键字的难度和工具需求预测
            if "complex calculation" in prompt.lower() or "data analysis" in prompt.lower():
                difficulty_score += 0.3
                required_tools.append("calculator_tool")
                required_tools.append("data_plotter_tool")
            if "write code" in prompt.lower():
                difficulty_score += 0.2
                required_tools.append("code_interpreter_tool")
            if "search internet" in prompt.lower() or "latest news" in prompt.lower():
                required_tools.append("web_search_tool")
    
            # 检查所需工具是否可用
            missing_tools = [tool for tool in required_tools if tool not in self.available_tools]
            if missing_tools:
                confidence_score -= 0.4 * len(missing_tools)
                print(f"  Warning: Missing tools for this task: {missing_tools}")
    
            # 根据难度调整信心
            confidence_score -= (difficulty_score - 0.5) * 0.3
    
            return {
                "predicted_difficulty": max(0, min(1, difficulty_score)),
                "predicted_success_confidence": max(0, min(1, confidence_score)),
                "required_tools": required_tools,
                "missing_tools": missing_tools
            }
    
    # 智能体工作流
    def agent_workflow_pre_execution(prompt: str, agent_tools: list[str]):
        planner = AgentPrePlanner(agent_tools)
    
        # 1. 预执行评估
        planning_results = planner.predict_task_feasibility(prompt)
        print(f"Pre-Execution Planning Results: {planning_results}")
    
        if planning_results["predicted_success_confidence"] < 0.6 or planning_results["missing_tools"]:
            print("Predicted low success confidence or missing tools. Task might be difficult or impossible for current agent. Consider re-routing or human assistance.")
            return None, planning_results # 不执行任务
        else:
            print("Predicted high success confidence. Proceeding with task execution.")
            # 2. 实际任务执行 (此处省略,可以调用Post-Execution或In-Execution Agent)
            # executor = AgentTaskExecutor()
            # task_output = executor.execute_task(prompt)
            # return task_output, planning_results
            return "Task will be executed.", planning_results
    
    # 示例调用
    # available_agent_tools = ["web_search_tool", "calculator_tool"]
    # prompt1 = "Find the latest stock price of Google and calculate its 5-day moving average."
    # agent_workflow_pre_execution(prompt1, available_agent_tools)
    
    # prompt2 = "Write a Python function to perform a complex calculation on a dataset and visualize it."
    # agent_workflow_pre_execution(prompt2, available_agent_tools) # 此时会发现缺少 code_interpreter_tool 和 data_plotter_tool

这些架构模式并非互斥,而是可以相互结合,形成一个多层次、全方位的自评估体系。例如,一个智能体可以在任务开始前进行预测,在任务执行过程中进行监控,并在任务完成后进行最终评估。

四、 实施自评估指标的策略与类型

现在,我们深入探讨智能体可以采用的具体自评估指标类型及其实现策略。这些策略通常结合了启发式规则、内部模型、大型语言模型(LLMs)的零样本/少样本能力,以及与外部工具的交互。

1. 信心度评分 (Confidence Score)

描述: 智能体对其自身输出或决策正确性的主观概率估计。这是一个非常基础且重要的指标。

实现策略:

  • 基于输出特征: 例如,如果输出内容很短、包含大量问号或表示不确定的词语(如“可能”、“或许”),则信心度降低。
  • 基于内部状态: 如果智能体在检索信息时发现多个矛盾的来源,或者在推理过程中遇到多次回溯,则信心度降低。
  • 基于LLM的后处理: 使用一个专门的LLM提示,让其评估另一个LLM的输出,并以分数形式给出信心度。
  • 基于语义相似度: 将智能体的输出与预期的答案或高质量的示例进行向量相似度比较。

代码示例:LLM作为信心度评分器

这里我们利用LLM的理解和推理能力,让它充当一个元评估器。

import openai
import json
import os

# 假设已经配置了OPENAI_API_KEY环境变量
# openai.api_key = os.getenv("OPENAI_API_KEY")

class LLMService:
    def __init__(self, model_name: str = "gpt-4-0125-preview"): # 或 gpt-3.5-turbo-0125
        self.model_name = model_name

    def generate_text(self, prompt: str, max_tokens: int = 500, temperature: float = 0.7) -> str:
        try:
            response = openai.chat.completions.create(
                model=self.model_name,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens,
                temperature=temperature,
            )
            return response.choices[0].message.content.strip()
        except Exception as e:
            print(f"Error calling LLM: {e}")
            return "LLM_ERROR: Could not generate response."

class LLMBasedConfidenceScorer:
    def __init__(self, llm_service: LLMService):
        self.llm = llm_service

    def score_confidence(self, original_prompt: str, agent_output: str) -> dict:
        evaluation_prompt = f"""
        你是一个专业的评估者,请根据以下原始任务指令和智能体输出,评估智能体输出的信心度。
        你的评估应该是一个JSON对象,包含以下字段:
        - "confidence_score": 一个介于0.0(完全不确定)到1.0(非常确定)之间的浮点数。
        - "reasoning": 解释你给出这个分数的原因。

        原始任务指令:
        {original_prompt}

        智能体输出:
        {agent_output}

        请只输出JSON对象,不要包含任何额外文字。
        """

        print("  LLM Scorer: Requesting confidence score...")
        llm_response = self.llm.generate_text(evaluation_prompt, max_tokens=200, temperature=0.0)

        try:
            # 尝试解析JSON
            evaluation = json.loads(llm_response)
            if not isinstance(evaluation, dict) or "confidence_score" not in evaluation or "reasoning" not in evaluation:
                raise ValueError("Invalid JSON structure from LLM.")
            return evaluation
        except json.JSONDecodeError:
            print(f"  LLM Scorer: Failed to parse JSON: {llm_response}")
            return {"confidence_score": 0.0, "reasoning": "Failed to parse LLM's evaluation output."}
        except ValueError as e:
            print(f"  LLM Scorer: {e}")
            return {"confidence_score": 0.0, "reasoning": str(e)}

# 假设我们有一个核心任务的Agent
class ContentGeneratorAgent:
    def __init__(self, llm_service: LLMService):
        self.llm = llm_service

    def generate_content(self, topic: str) -> str:
        gen_prompt = f"请撰写一篇关于'{topic}'的短篇文章,内容应包含至少三个关键点。"
        print(f"  Content Generator: Generating content for '{topic}'...")
        return self.llm.generate_text(gen_prompt, max_tokens=300)

# 主程序流程
def run_confidence_evaluation_example():
    print("--- 信心度评分示例 ---")
    llm_service = LLMService()
    generator_agent = ContentGeneratorAgent(llm_service)
    confidence_scorer = LLMBasedConfidenceScorer(llm_service)

    # 场景1: 正常输出
    topic_good = "可持续发展的重要性"
    original_prompt_good = f"撰写一篇关于'{topic_good}'的短篇文章,内容应包含至少三个关键点。"
    agent_output_good = generator_agent.generate_content(original_prompt_good)
    print(f"nAgent Output (Good): {agent_output_good}")
    confidence_score_good = confidence_scorer.score_confidence(original_prompt_good, agent_output_good)
    print(f"Confidence Score (Good): {confidence_score_good}")

    # 场景2: 故意制造一个模糊或不确定的输出
    topic_uncertain = "关于未来火星移民计划的未知风险"
    original_prompt_uncertain = f"撰写一篇关于'{topic_uncertain}'的短篇文章,内容应包含至少三个关键点。"
    # 模拟一个LLM在处理不确定话题时可能产生的模糊输出
    agent_output_uncertain = generator_agent.llm.generate_text(f"请撰写一篇关于'{topic_uncertain}'的短篇文章,内容应包含至少三个关键点。请在文中加入一些不确定性的表达。", max_tokens=300)
    print(f"nAgent Output (Uncertain): {agent_output_uncertain}")
    confidence_score_uncertain = confidence_scorer.score_confidence(original_prompt_uncertain, agent_output_uncertain)
    print(f"Confidence Score (Uncertain): {confidence_score_uncertain}")

# run_confidence_evaluation_example()

2. 完整性/覆盖度评分 (Completeness/Coverage Score)

描述: 评估智能体输出是否包含了所有必需的信息,是否满足了任务指令中提出的所有要求。

实现策略:

  • 规则匹配: 检查输出中是否包含特定的关键字、段落结构或数据点。
  • 任务分解与检查: 如果原始任务被分解为子任务,则检查每个子任务是否都已完成并集成到最终输出中。
  • LLM辅助检查: 使用LLM对比任务指令和输出,列出未满足的要求。
  • 语义检查: 比较输出的语义内容与任务要求中提到的关键概念。

代码示例:LLM辅助完整性检查

class LLMBasedCompletenessScorer:
    def __init__(self, llm_service: LLMService):
        self.llm = llm_service

    def score_completeness(self, original_prompt: str, agent_output: str) -> dict:
        evaluation_prompt = f"""
        你是一个严谨的评估者,请根据以下原始任务指令和智能体输出,评估智能体输出的完整性。
        你的评估应该是一个JSON对象,包含以下字段:
        - "completeness_score": 一个介于0.0(完全不完整)到1.0(完全完整)之间的浮点数。
        - "missing_elements": 一个字符串列表,列出智能体输出中未能满足原始指令的任何具体要求或缺失的信息。如果没有缺失,则为空列表。
        - "reasoning": 解释你给出这个分数的原因。

        原始任务指令:
        {original_prompt}

        智能体输出:
        {agent_output}

        请只输出JSON对象,不要包含任何额外文字。
        """

        print("  LLM Scorer: Requesting completeness score...")
        llm_response = self.llm.generate_text(evaluation_prompt, max_tokens=300, temperature=0.0)

        try:
            evaluation = json.loads(llm_response)
            if not isinstance(evaluation, dict) or "completeness_score" not in evaluation or "missing_elements" not in evaluation:
                raise ValueError("Invalid JSON structure from LLM.")
            return evaluation
        except json.JSONDecodeError:
            print(f"  LLM Scorer: Failed to parse JSON: {llm_response}")
            return {"completeness_score": 0.0, "missing_elements": ["JSON_PARSE_ERROR"], "reasoning": "Failed to parse LLM's evaluation output."}
        except ValueError as e:
            print(f"  LLM Scorer: {e}")
            return {"completeness_score": 0.0, "missing_elements": [str(e)], "reasoning": str(e)}

def run_completeness_evaluation_example():
    print("n--- 完整性/覆盖度评分示例 ---")
    llm_service = LLMService()
    generator_agent = ContentGeneratorAgent(llm_service) # 复用ContentGeneratorAgent
    completeness_scorer = LLMBasedCompletenessScorer(llm_service)

    # 场景1: 完整满足要求
    topic_complete = "量子计算的基本原理"
    original_prompt_complete = f"请撰写一篇关于'{topic_complete}'的短篇文章,内容应包含至少三个关键点:叠加、纠缠和量子比特。文章长度至少150字。"
    # 模拟一个满足所有要求的输出
    agent_output_complete = generator_agent.llm.generate_text(original_prompt_complete, max_tokens=400)
    print(f"nAgent Output (Complete): {agent_output_complete}")
    completeness_score_complete = completeness_scorer.score_completeness(original_prompt_complete, agent_output_complete)
    print(f"Completeness Score (Complete): {completeness_score_complete}")

    # 场景2: 缺失部分要求
    topic_incomplete = "人工智能在医疗领域的应用"
    original_prompt_incomplete = f"请撰写一篇关于'{topic_incomplete}'的短篇文章,内容应包含至少四个关键点:诊断辅助、药物研发、个性化治疗和远程医疗。文章长度至少200字。"
    # 模拟一个故意遗漏一个关键点的输出
    agent_output_incomplete = generator_agent.llm.generate_text(
        f"请撰写一篇关于'{topic_incomplete}'的短篇文章,内容应包含诊断辅助、药物研发和个性化治疗这三个关键点。文章长度至少200字。", max_tokens=400
    )
    print(f"nAgent Output (Incomplete): {agent_output_incomplete}")
    completeness_score_incomplete = completeness_scorer.score_completeness(original_prompt_incomplete, agent_output_incomplete)
    print(f"Completeness Score (Incomplete): {completeness_score_incomplete}")

# run_completeness_evaluation_example()

3. 相关性/对齐度评分 (Relevance/Alignment Score)

描述: 评估智能体输出与原始任务指令或用户意图的匹配程度,是否偏离主题。

实现策略:

  • 关键词匹配: 检查输出中是否包含原始指令中的核心关键词。
  • 语义相似度: 使用文本嵌入(如BERT, Sentence-BERT)计算输出与指令之间的余弦相似度。
  • LLM判断: 让LLM判断输出是否紧密围绕主题,是否有跑题内容。

代码示例:嵌入式语义相似度评分

from sentence_transformers import SentenceTransformer, util
import torch

class EmbeddingService:
    def __init__(self, model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2'):
        self.model = SentenceTransformer(model_name)

    def get_embedding(self, text: str) -> torch.Tensor:
        return self.model.encode(text, convert_to_tensor=True)

class EmbeddingBasedRelevanceScorer:
    def __init__(self, embedding_service: EmbeddingService):
        self.embedding_service = embedding_service

    def score_relevance(self, original_prompt: str, agent_output: str) -> dict:
        print("  Embedding Scorer: Calculating relevance score...")
        prompt_embedding = self.embedding_service.get_embedding(original_prompt)
        output_embedding = self.embedding_service.get_embedding(agent_output)

        # 计算余弦相似度
        cosine_similarity = util.cos_sim(prompt_embedding, output_embedding).item()

        # 将相似度映射到0-1范围,作为相关性分数
        # 余弦相似度本身就在-1到1,但对于文本通常在0到1之间,0.5以下可能已经很差了
        relevance_score = (cosine_similarity + 1) / 2 # 映射到0-1,-1->0, 0->0.5, 1->1

        return {
            "relevance_score": relevance_score,
            "cosine_similarity": cosine_similarity,
            "reasoning": "Calculated based on semantic similarity between prompt and output embeddings."
        }

def run_relevance_evaluation_example():
    print("n--- 相关性/对齐度评分示例 ---")
    # 注意:SentenceTransformer模型下载可能需要时间
    embedding_service = EmbeddingService()
    relevance_scorer = EmbeddingBasedRelevanceScorer(embedding_service)
    llm_service = LLMService()
    generator_agent = ContentGeneratorAgent(llm_service) # 复用ContentGeneratorAgent

    # 场景1: 高度相关
    topic_relevant = "如何有效学习一门新编程语言"
    original_prompt_relevant = f"撰写一篇关于'{topic_relevant}'的短篇文章,给出学习新编程语言的实用建议。"
    agent_output_relevant = generator_agent.llm.generate_text(original_prompt_relevant, max_tokens=300)
    print(f"nAgent Output (Relevant): {agent_output_relevant}")
    relevance_score_relevant = relevance_scorer.score_relevance(original_prompt_relevant, agent_output_relevant)
    print(f"Relevance Score (Relevant): {relevance_score_relevant}")

    # 场景2: 相关性较低(跑题)
    topic_off_topic = "Python编程的最佳实践"
    original_prompt_off_topic = f"撰写一篇关于'{topic_off_topic}'的短篇文章,聚焦于Python代码风格和模块化。"
    # 模拟一个跑题的输出,例如写成了Python的历史或应用场景
    agent_output_off_topic = generator_agent.llm.generate_text(
        "请撰写一篇关于Python编程语言历史与广泛应用的短文章。", max_tokens=300
    )
    print(f"nAgent Output (Off-topic): {agent_output_off_topic}")
    relevance_score_off_topic = relevance_scorer.score_relevance(original_prompt_off_topic, agent_output_off_topic)
    print(f"Relevance Score (Off-topic): {relevance_score_off_topic}")

# run_relevance_evaluation_example()

4. 约束遵守/安全性评分 (Constraint Adherence/Safety Score)

描述: 评估智能体输出是否遵守了所有明确的约束条件(如格式要求、长度限制、禁止词语)以及是否符合安全和伦理准则(如无偏见、无有害内容)。

实现策略:

  • 正则表达式/关键词过滤: 对输出进行模式匹配,检查敏感词、格式错误等。
  • 规则引擎: 构建复杂的规则集来验证输出是否满足一系列约束。
  • LLM辅助审查: 让LLM根据安全准则和约束条件对输出进行审查。
  • 外部安全API: 集成内容审查或偏见检测API。

代码示例:基于规则和LLM的约束遵守评分

class ConstraintAdherenceScorer:
    def __init__(self, llm_service: LLMService, prohibited_keywords: list[str]):
        self.llm = llm_service
        self.prohibited_keywords = [k.lower() for k in prohibited_keywords]

    def score_adherence(self, original_prompt: str, agent_output: str, max_length: int = 500) -> dict:
        print("  Constraint Scorer: Checking adherence to constraints...")

        violations = []
        score = 1.0 # 初始满分

        # 规则1: 长度限制
        if len(agent_output) > max_length:
            violations.append(f"Output exceeds maximum length of {max_length} characters.")
            score -= 0.3

        # 规则2: 禁止关键词
        for keyword in self.prohibited_keywords:
            if keyword in agent_output.lower():
                violations.append(f"Output contains prohibited keyword: '{keyword}'.")
                score -= 0.5 # 严重违规

        # 规则3: LLM辅助检查更复杂的约束(例如,情感倾向,或是否包含特定类型的信息)
        # 我们可以让LLM判断输出是否包含“争议性观点”
        llm_check_prompt = f"""
        你是一个内容审查员,请评估以下智能体输出是否包含任何争议性、冒犯性或不恰当的内容。
        你的评估应该是一个JSON对象,包含以下字段:
        - "is_safe": 布尔值,如果内容安全则为true,否则为false。
        - "safety_concerns": 字符串列表,列出任何发现的安全问题。如果没有,则为空列表。
        - "reasoning": 解释你的判断。

        智能体输出:
        {agent_output}

        请只输出JSON对象,不要包含任何额外文字。
        """
        llm_safety_response = self.llm.generate_text(llm_check_prompt, max_tokens=150, temperature=0.0)
        try:
            safety_eval = json.loads(llm_safety_response)
            if not safety_eval.get("is_safe", True):
                violations.extend(safety_eval.get("safety_concerns", []))
                score -= 0.8 # 安全问题是严重的
        except json.JSONDecodeError:
            print(f"  Constraint Scorer: Failed to parse LLM safety check: {llm_safety_response}")
            violations.append("Failed to perform LLM safety check.")
            score -= 0.1 # 无法检查也是个问题

        return {
            "adherence_score": max(0.0, score),
            "violations": violations,
            "reasoning": "Combined rule-based and LLM-based constraint checks."
        }

def run_constraint_evaluation_example():
    print("n--- 约束遵守/安全性评分示例 ---")
    llm_service = LLMService()
    generator_agent = ContentGeneratorAgent(llm_service) # 复用ContentGeneratorAgent
    prohibited_words = ["暴力", "仇恨", "歧视", "非法"]
    constraint_scorer = ConstraintAdherenceScorer(llm_service, prohibited_words)

    # 场景1: 遵守所有约束
    topic_compliant = "如何在工作中保持专注"
    original_prompt_compliant = f"撰写一篇关于'{topic_compliant}'的短篇文章,长度不超过200字,不包含任何负面词语。"
    agent_output_compliant = generator_agent.llm.generate_text(
        "如何在工作中保持专注?以下是一些积极的方法:番茄工作法、减少干扰、定期休息、设定清晰目标。这些能帮助你提高效率并享受工作过程。", max_tokens=200
    )
    print(f"nAgent Output (Compliant): {agent_output_compliant}")
    adherence_score_compliant = constraint_scorer.score_adherence(original_prompt_compliant, agent_output_compliant, max_length=200)
    print(f"Adherence Score (Compliant): {adherence_score_compliant}")

    # 场景2: 违反长度约束和含有禁止词语
    topic_violation = "关于历史冲突的分析"
    original_prompt_violation = f"撰写一篇关于'{topic_violation}'的短篇文章,长度不超过100字,不包含任何暴力或歧视性词语。"
    # 模拟一个超出长度且包含禁止词的输出
    agent_output_violation = generator_agent.llm.generate_text(
        "关于历史冲突的分析是一个复杂的话题。回顾历史,我们不难发现许多关于暴力和仇恨的事件。这些事件往往由根深蒂固的歧视观念引发,导致了社会的动荡和人民的苦难。我们需要深入研究这些冲突的根源,以期在未来避免重蹈覆辙,促进和平与理解。这是一个漫长而艰巨的任务,需要全球的共同努力和持续的对话。", max_tokens=300
    )
    print(f"nAgent Output (Violation): {agent_output_violation}")
    adherence_score_violation = constraint_scorer.score_adherence(original_prompt_violation, agent_output_violation, max_length=100)
    print(f"Adherence Score (Violation): {adherence_score_violation}")

# run_constraint_evaluation_example()

5. 效率/资源使用评分 (Efficiency/Resource Usage Score)

描述: 评估智能体在执行任务时所消耗的资源,例如时间、计算量(如LLM的tokens数量)、API调用次数等。

实现策略:

  • 计时器: 记录任务执行的开始和结束时间。
  • API日志: 统计LLM或其他外部API的调用次数和tokens消耗。
  • 自定义计数器: 记录内部迭代次数、搜索深度等。

代码示例:简单的效率评分

import time

class EfficiencyScorer:
    def score_efficiency(self, start_time: float, end_time: float, llm_tokens_used: int = 0, api_calls: int = 0) -> dict:
        duration = end_time - start_time
        efficiency_score = 1.0 # 初始满分
        feedback = []

        # 规则1: 时间效率
        if duration > 10.0: # 假设10秒是阈值
            efficiency_score -= 0.4
            feedback.append(f"Task took too long: {duration:.2f} seconds.")
        elif duration > 5.0:
            efficiency_score -= 0.1
            feedback.append(f"Task took {duration:.2f} seconds, moderately slow.")

        # 规则2: LLM Token消耗
        if llm_tokens_used > 1000: # 假设1000 tokens是阈值
            efficiency_score -= 0.3
            feedback.append(f"High LLM token usage: {llm_tokens_used} tokens.")
        elif llm_tokens_used > 500:
            efficiency_score -= 0.1
            feedback.append(f"Moderate LLM token usage: {llm_tokens_used} tokens.")

        # 规则3: API调用次数
        if api_calls > 5: # 假设5次是阈值
            efficiency_score -= 0.2
            feedback.append(f"Excessive API calls: {api_calls} calls.")

        return {
            "efficiency_score": max(0.0, efficiency_score),
            "duration_seconds": duration,
            "llm_tokens_used": llm_tokens_used,
            "api_calls": api_calls,
            "feedback": feedback
        }

def run_efficiency_evaluation_example():
    print("n--- 效率/资源使用评分示例 ---")
    efficiency_scorer = EfficiencyScorer()

    # 模拟一个任务执行
    start_time = time.time()
    # 假设LLM调用了,并消耗了tokens
    # 假设任务执行了某些API调用
    time.sleep(7) # 模拟任务执行时间
    end_time = time.time()

    # 假设LLM使用了800 tokens,进行了3次API调用
    efficiency_results_moderate = efficiency_scorer.score_efficiency(start_time, end_time, llm_tokens_used=800, api_calls=3)
    print(f"Efficiency Results (Moderate): {efficiency_results_moderate}")

    start_time_slow = time.time()
    time.sleep(12) # 模拟更长的任务执行时间
    end_time_slow = time.time()
    # 假设LLM使用了1200 tokens,进行了6次API调用
    efficiency_results_low = efficiency_scorer.score_efficiency(start_time_slow, end_time_slow, llm_tokens_used=1200, api_calls=6)
    print(f"nEfficiency Results (Low): {efficiency_results_low}")

# run_efficiency_evaluation_example()

6. 事实准确性/真实性评分 (Factual Accuracy/Truthfulness Score)

描述: 评估智能体输出中包含的事实信息是否准确无误。

实现策略:

  • 知识库查询: 将输出中的关键事实提取出来,与内部知识库、外部数据库(如维基百科API、Google Knowledge Graph)进行比对。
  • 搜索引擎验证: 使用搜索引擎验证输出中的声明,检查是否有支持或反驳的证据。
  • LLM辅助验证: 训练或提示LLM作为“事实核查员”,但需要注意LLM本身可能存在幻觉。通常需要结合外部工具。

代码示例:工具辅助的事实准确性评分

这个例子需要一个模拟的外部知识库或API来验证事实。


class ExternalKnowledgeBase:
    def get_fact(self, query: str) -> str:
        # 模拟一个简单的知识库查询
        facts = {
            "巴黎是法国的首都": "True",
            "地球是平的": "False",
            "水在100摄氏度沸腾": "True",
            "人工智能是人类创造的": "True",
            "太阳系有九大行星": "False (冥王星已降级,通常认为是八大行星)",
            "Python是一种编程语言": "True"
        }
        return facts.get(query, "Unknown")

class FactualAccuracyScorer:
    def __init__(self, llm_service: LLMService, knowledge_base: ExternalKnowledgeBase):
        self.llm = llm_service
        self.kb = knowledge_base

    def score_accuracy(self, agent_output: str) -> dict:
        print("  Accuracy Scorer: Extracting facts for verification...")

        # 使用LLM提取输出中的关键事实陈述
        fact_extraction_prompt = f"""
        请从以下文本中提取所有独立的事实陈述,每条陈述一行。
        例如:
        文本:地球是圆的。太阳会发光。
        输出:
        地球是圆的
        太阳会发光

        文本:
        {agent_output}

        输出:
        """
        facts_raw = self.llm.generate_text(fact_extraction_prompt, max_tokens=200, temperature=0.0)
        facts_to_verify = [f.strip() for f in facts_raw.split('n') if f.strip()]

        verified_

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注