Agent 평가

from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, Field

class EvalResult(BaseModel):
    score: int = Field(ge=1, le=5, description="1~5 점수")
    factuality: bool   = Field(description="사실 정확성")
    completeness: bool = Field(description="요청 사항 완전 처리 여부")
    conciseness: bool  = Field(description="불필요한 내용 없음")
    reasoning: str     = Field(description="채점 근거 (1~2 문장)")

judge_llm = ChatAnthropic(model="claude-opus-4-7")

async def evaluate_output(
    question: str,
    agent_output: str,
    reference_answer: str = ""
) -> EvalResult:
    """Agent 최종 출력을 LLM-as-Judge로 평가"""
    structured_llm = judge_llm.with_structured_output(EvalResult)

    reference_section = (
        f"\n참조 답변:\n{reference_answer}"
        if reference_answer else ""
    )

    result = await structured_llm.ainvoke([{
        "role": "user",
        "content": f"""AI Agent의 응답을 평가하세요.

질문: {question}
Agent 응답: {agent_output}{reference_section}

평가 기준:
5점: 완벽히 정확하고 완전하며 간결
4점: 대체로 정확, 사소한 누락
3점: 부분적으로 정확하거나 불완전
2점: 부정확하거나 질문 오해
1점: 완전히 부적절"""
    }])
    return result

from dataclasses import dataclass
from langchain_core.messages import AIMessage, ToolMessage

@dataclass
class TrajectoryStep:
    thought: str
    tool_name: str
    tool_args: dict
    tool_result: str

def extract_trajectory(messages: list) -> list[TrajectoryStep]:
    """메시지 이력에서 도구 호출 경로 추출"""
    steps = []
    for i, msg in enumerate(messages):
        if isinstance(msg, AIMessage) and msg.tool_calls:
            for tc in msg.tool_calls:
                # 다음 ToolMessage에서 결과 찾기
                result = ""
                for next_msg in messages[i+1:]:
                    if (isinstance(next_msg, ToolMessage)
                            and next_msg.tool_call_id == tc["id"]):
                        result = next_msg.content
                        break
                steps.append(TrajectoryStep(
                    thought=msg.content or "",
                    tool_name=tc["name"],
                    tool_args=tc["args"],
                    tool_result=result
                ))
    return steps

class TrajectoryScore(BaseModel):
    efficiency: int   = Field(ge=1, le=5, description="불필요한 도구 호출 없음")
    correctness: int  = Field(ge=1, le=5, description="올바른 도구 선택")
    order: int        = Field(ge=1, le=5, description="논리적 실행 순서")
    issues: list[str] = Field(description="발견된 문제점 목록")

async def evaluate_trajectory(
    question: str,
    trajectory: list[TrajectoryStep],
    expected_tools: list[str] = []
) -> TrajectoryScore:
    traj_str = "\n".join(
        f"Step {i+1}: {s.tool_name}({s.tool_args}) → {s.tool_result[:200]}"
        for i, s in enumerate(trajectory)
    )
    expected_str = ", ".join(expected_tools) if expected_tools else "명시되지 않음"

    structured_llm = judge_llm.with_structured_output(TrajectoryScore)
    return await structured_llm.ainvoke([{
        "role": "user",
        "content": f"""Agent의 도구 사용 경로를 평가하세요.

질문: {question}
기대 도구: {expected_str}
실제 경로:
{traj_str}

중복 호출, 불필요한 도구, 잘못된 순서를 찾아 점수를 매기세요."""
    }])

from langsmith import Client
from langsmith.evaluation import evaluate, LangChainStringEvaluator
import os

client = Client()  # LANGCHAIN_API_KEY 환경변수 필요

# 테스트 데이터셋 생성
dataset = client.create_dataset(
    "agent-eval-v1",
    description="Agent 성능 평가 데이터셋"
)

test_cases = [
    {
        "inputs": {"question": "2025년 AI 트렌드를 검색하고 요약해줘"},
        "outputs": {"expected_tools": ["web_search"]},
    },
    {
        "inputs": {"question": "피보나치 수열 n=20을 Python으로 계산해줘"},
        "outputs": {"expected_tools": ["python_repl"]},
    },
]
client.create_examples(
    inputs=[tc["inputs"] for tc in test_cases],
    outputs=[tc["outputs"] for tc in test_cases],
    dataset_id=dataset.id
)

# 커스텀 평가 함수
def tool_accuracy_evaluator(run, example) -> dict:
    """실제 사용 도구 vs 기대 도구 일치 여부"""
    outputs = run.outputs or {}
    used_tools = outputs.get("tools_used", [])
    expected = example.outputs.get("expected_tools", [])

    overlap = set(used_tools) & set(expected)
    score = len(overlap) / len(expected) if expected else 1.0

    return {"key": "tool_accuracy", "score": score}

# 평가 실행
results = evaluate(
    lambda inputs: agent_graph.invoke(inputs),
    data="agent-eval-v1",
    evaluators=[
        tool_accuracy_evaluator,
        LangChainStringEvaluator("criteria", config={
            "criteria": {
                "helpfulness": "응답이 질문에 실질적으로 도움이 되는가?",
                "accuracy":     "사실 관계가 정확한가?",
            }
        })
    ],
    experiment_prefix="claude-opus-4-7-v1",
    num_repetitions=3  # 각 케이스 3회 반복으로 변동성 측정
)
print(results.to_pandas()[["tool_accuracy", "helpfulness", "accuracy"]].describe())