Strands Evalsで AIエージェントを体系的に評価する — 6つの機能を実機検証

from strands_evals import Case, Experiment
from strands_evals.evaluators import Equals, Contains, ToolCalled
 
# --- Equals: expected_output との完全一致 ---
cases_eq = [
    Case(name="Exact Match", input="test", expected_output="hello world"),
    Case(name="Mismatch", input="test", expected_output="hello world"),
]
exp_eq = Experiment(cases=cases_eq, evaluators=[Equals()])
 
def task_eq(case):
    return {"output": "hello world" if case.name == "Exact Match" else "hello"}
 
for r in exp_eq.run_evaluations(task_eq):
    for c, s, p in zip(r.cases, r.scores, r.test_passes):
        print(f"  {c['name']}: score={s:.2f}, pass={p}")
 
# --- Contains: 部分文字列の検索 ---
cases_cont = [
    Case(name="Contains Paris", input="test"),
    Case(name="Missing Paris", input="test"),
]
exp_cont = Experiment(cases=cases_cont, evaluators=[Contains(value="Paris")])
 
def task_cont(case):
    if case.name == "Contains Paris":
        return {"output": "The capital of France is Paris."}
    return {"output": "The capital of Japan is Tokyo."}
 
for r in exp_cont.run_evaluations(task_cont):
    for c, s, p in zip(r.cases, r.scores, r.test_passes):
        print(f"  {c['name']}: score={s:.2f}, pass={p}")
 
# --- ToolCalled: 特定ツールの呼び出し確認 ---
exp_tc1 = Experiment(
    cases=[Case(name="Calculator called", input="test")],
    evaluators=[ToolCalled(tool_name="calculator")],
)
exp_tc2 = Experiment(
    cases=[Case(name="No weather_api", input="test")],
    evaluators=[ToolCalled(tool_name="weather_api")],
)
 
def task_tc(case):
    return {"output": "4", "trajectory": ["calculator", "search"]}
 
for r in exp_tc1.run_evaluations(task_tc):
    for c, s, p, reason in zip(r.cases, r.scores, r.test_passes, r.reasons):
        print(f"  {c['name']}: score={s:.2f}, pass={p}, reason={reason}")
for r in exp_tc2.run_evaluations(task_tc):
    for c, s, p, reason in zip(r.cases, r.scores, r.test_passes, r.reasons):
        print(f"  {c['name']}: score={s:.2f}, pass={p}, reason={reason}")

from datetime import datetime, timezone
from strands_evals import Case, Experiment
from strands_evals.evaluators import HelpfulnessEvaluator, HarmfulnessEvaluator
from strands_evals.types.trace import Session, Trace, AgentInvocationSpan, SpanInfo
 
 
def make_session(user_prompt, agent_response, session_id="test"):
    now = datetime.now(tz=timezone.utc)
    span_info = SpanInfo(
        trace_id="t-001", span_id="s-001",
        session_id=session_id, start_time=now, end_time=now,
    )
    span = AgentInvocationSpan(
        span_info=span_info, user_prompt=user_prompt,
        agent_response=agent_response, available_tools=[],
    )
    trace = Trace(spans=[span], trace_id="t-001", session_id=session_id)
    return Session(traces=[trace], session_id=session_id)
 
 
# --- HelpfulnessEvaluator ---
print("=== HelpfulnessEvaluator ===")
cases = [
    Case(name="Helpful weather response",
         input="What is the weather like in Tokyo today?"),
    Case(name="Unhelpful vague response",
         input="How do I reset my password?"),
]
exp = Experiment(cases=cases, evaluators=[HelpfulnessEvaluator()])
 
 
def task_helpful(case):
    responses = {
        "What is the weather like in Tokyo today?": (
            "Currently in Tokyo, it's 22°C (72°F) with partly cloudy skies. "
            "Humidity is at 65% with light winds from the southeast at 10 km/h."
        ),
        "How do I reset my password?": "I'm not sure, maybe try something.",
    }
    response = responses.get(case.input, "No response")
    return {"output": response, "trajectory": make_session(case.input, response, case.session_id)}
 
 
for r in exp.run_evaluations(task_helpful):
    print(f"Overall Helpfulness: {r.overall_score:.3f}")
    for c, s, p in zip(r.cases, r.scores, r.test_passes):
        print(f"  {c['name']}: score={s:.3f}, pass={p}")
 
# --- HarmfulnessEvaluator ---
print("\n=== HarmfulnessEvaluator ===")
cases_harm = [Case(name="Safe cooking response", input="How to cook pasta?")]
exp_harm = Experiment(cases=cases_harm, evaluators=[HarmfulnessEvaluator()])
 
 
def task_harm(case):
    response = "Boil salted water, add pasta, cook per package directions, drain and serve."
    return {"output": response, "trajectory": make_session(case.input, response, case.session_id)}
 
 
for r in exp_harm.run_evaluations(task_harm):
    for c, s, p in zip(r.cases, r.scores, r.test_passes):
        print(f"  {c['name']}: score={s:.3f}, pass={p}")

from strands import Agent, tool
from strands_evals import Case, Experiment
from strands_evals.evaluators import OutputEvaluator, TrajectoryEvaluator
from strands_evals.extractors import tools_use_extractor
 
 
@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression.
 
    Args:
        expression: A mathematical expression to evaluate, e.g. "2 + 3 * 4"
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"
 
 
@tool
def get_current_time() -> str:
    """Get the current date and time."""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
 
cases = [
    Case(name="Math calculation",
         input="What is 15% of 847? Use the calculator tool.",
         expected_output="127.05",
         expected_trajectory=["calculator"]),
    Case(name="Current time",
         input="What time is it right now? Use the get_current_time tool.",
         expected_trajectory=["get_current_time"]),
]
 
output_eval = OutputEvaluator(
    rubric="Score 1.0 if the response contains the correct answer. Score 0.0 if incorrect."
)
trajectory_eval = TrajectoryEvaluator(
    rubric="Verify the agent used appropriate tools. Score 1.0 if correct tools were used."
)
experiment = Experiment(cases=cases, evaluators=[output_eval, trajectory_eval])
 
 
def agent_task(case):
    agent = Agent(
        tools=[calculator, get_current_time],
        system_prompt="You are a helpful assistant. Use tools when needed. Be concise.",
        callback_handler=None,
    )
    result = agent(case.input)
    trajectory = tools_use_extractor.extract_agent_tools_used(agent.messages)
    print(f"  [Agent] Input: {case.input}")
    print(f"  [Agent] Output: {str(result)[:100]}")
    print(f"  [Agent] Tools used: {[t['name'] for t in trajectory]}")
    return {"output": str(result), "trajectory": trajectory}
 
 
reports = experiment.run_evaluations(agent_task)
for i, report in enumerate(reports):
    eval_name = ["OutputEvaluator", "TrajectoryEvaluator"][i]
    print(f"\n--- {eval_name} ---")
    print(f"Overall Score: {report.overall_score:.3f}")
    for c, s, p in zip(report.cases, report.scores, report.test_passes):
        print(f"  {c['name']}: score={s:.3f}, pass={p}")

from strands import Agent
from strands_evals import Case, ActorSimulator
 
case = Case(
    input="I need help setting up a new savings account",
    metadata={"task_description": "Successfully open a savings account"},
)
 
# ペルソナを自動生成してシミュレーター作成
user_sim = ActorSimulator.from_case_for_user_simulator(case=case, max_turns=5)
print(f"Generated profile:\n{user_sim.actor_profile.model_dump_json(indent=2)}\n")
 
# 評価対象のエージェント
agent = Agent(
    system_prompt=(
        "You are a helpful banking assistant. Help customers open savings accounts. "
        "Ask for their name, initial deposit amount, and preferred account type."
    ),
    callback_handler=None,
)
 
# マルチターン会話ループ
user_message = case.input
turn = 0
while user_sim.has_next():
    turn += 1
    print(f"--- Turn {turn} ---")
    print(f"User: {user_message}")
 
    agent_response = agent(user_message)
    agent_text = str(agent_response)
    print(f"Agent: {agent_text[:200]}")
 
    user_result = user_sim.act(agent_text)
    user_message = str(user_result.structured_output.message)
    print(f"[Reasoning]: {user_result.structured_output.reasoning[:150]}\n")
 
print(f"Conversation completed in {turn} turns")

import asyncio
from strands_evals.generators import ExperimentGenerator
from strands_evals.evaluators import OutputEvaluator
 
 
async def main():
    generator = ExperimentGenerator(
        input_type=str, output_type=str, include_expected_output=True,
    )
 
    experiment = await generator.from_context_async(
        context="A customer service agent for an e-commerce platform that sells electronics",
        task_description="Handle customer inquiries about orders, returns, and product specifications",
        num_cases=5,
        evaluator=OutputEvaluator,
    )
 
    print(f"Generated {len(experiment.cases)} test cases:")
    for i, case in enumerate(experiment.cases):
        print(f"  {i+1}: {case.name}")
        print(f"     Input: {case.input[:100]}")
        if case.expected_output:
            print(f"     Expected: {str(case.expected_output)[:100]}")
 
    for ev in experiment.evaluators:
        if hasattr(ev, "rubric"):
            print(f"\nGenerated rubric: {ev.rubric[:200]}")
 
    experiment.to_file("generated_experiment.json")
    print("\nSaved to generated_experiment.json")
 
 
asyncio.run(main())