주요 개념
- 에이전트 평가(Agent evaluation)
- 평가자(Evaluators)
- LLM-as-judge 평가자
이 튜토리얼에서는 사용자가 디지털 음악 스토어를 탐색할 수 있도록 돕는 고객 지원 챗봇을 구축합니다. 그리고 챗봇에 대해 실행할 수 있는 세 가지 가장 효과적인 평가 유형을 살펴봅니다.
- 최종 응답 평가: 에이전트의 최종 응답을 평가합니다.
- 경로(trajectory) 평가: 에이전트가 최종 답변에 도달하기 위해 예상한 경로(예: 도구 호출 순서)를 따랐는지 평가합니다.
- 단일 단계 평가: 에이전트의 특정 단계를 개별적으로 평가합니다(예: 주어진 입력에 대해 적절한 첫 번째 도구를 선택했는지 등).
이 튜토리얼에서는 LangGraph를 사용하여 에이전트를 구축하지만, 여기서 소개하는 기법과 LangSmith 기능은 프레임워크에 구애받지 않습니다.
환경 설정
환경 구성
필요한 의존성을 설치합니다.
pip install -U langgraph langchain[openai]OpenAI와 LangSmith를 위한 환경 변수를 설정합니다.
import getpass
import os
def _set_env(var: str) -> None:
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"Set {var}: ")
os.environ["LANGSMITH_TRACING"] = "true"
_set_env("LANGSMITH_API_KEY")
_set_env("OPENAI_API_KEY")데이터베이스 다운로드
이 튜토리얼에서는 SQLite 데이터베이스를 사용합니다. SQLite는 간단하게 사용할 수 있는 경량 데이터베이스입니다.
예시로 디지털 미디어 스토어를 나타내는 샘플 데이터베이스인 chinook를 사용합니다.
아래 코드를 통해 데이터베이스를 다운로드할 수 있습니다.
import requests
url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
response = requests.get(url)
if response.status_code == 200:
with open("chinook.db", "wb") as file:
file.write(response.content)
print("File downloaded and saved as Chinook.db")
else:
print(f"Failed to download the file. Status code: {response.status_code}")데이터베이스의 일부 예시 데이터:
import sqlite3
# ... (생략)
[(1, 'AC/DC'), (2, 'Accept'), (3, 'Aerosmith'), ...]데이터베이스 스키마는 여기에서 확인할 수 있습니다.
고객 지원 에이전트 정의
LangGraph 에이전트를 생성합니다. 이 에이전트는 데이터베이스에 제한적으로 접근할 수 있습니다.
데모 목적상, 에이전트는 두 가지 기본 요청을 지원합니다.
- 조회(Lookup): 고객이 곡 제목, 아티스트 이름, 앨범을 조회할 수 있습니다.
예: "Jimi Hendrix의 곡이 있나요?" - 환불(Refund): 고객이 과거 구매에 대해 환불을 요청할 수 있습니다.
예: "Claude Shannon인데, 지난주에 구매한 것 환불해줄 수 있나요?"
데모의 단순화를 위해 환불은 해당 데이터베이스 레코드를 삭제하는 방식으로 처리합니다.
실제 서비스에서는 사용자 인증 등 보안 조치가 필요합니다.
에이전트의 로직은 두 개의 서브그래프(조회, 환불)와 이를 라우팅하는 상위 그래프로 구성됩니다.
환불 에이전트
환불 처리 에이전트는 다음을 수행해야 합니다.
- 데이터베이스에서 고객의 구매 기록을 찾기
- 환불을 위해 관련 Invoice 및 InvoiceLine 레코드 삭제
테스트를 쉽게 하기 위해 "mock" 모드를 추가하여 실제 데이터 변경 없이 동작을 시뮬레이션할 수 있습니다.
import sqlite3
def _refund(invoice_id: int | None, invoice_line_ids: list[int] | None, mock: bool = False) -> float:
# ... (구현 생략)
def _lookup(...):
# ... (구현 생략)그래프는 다음과 같이 세 가지 주요 경로로 구성됩니다.
- 대화에서 고객 및 구매 정보 추출
- 아래 세 경로 중 하나로 라우팅
- 환불 경로: 환불 처리에 충분한 구매 정보가 있을 때
- 조회 경로: 고객 정보(이름, 전화번호 등)로 구매 내역을 조회할 수 있을 때
- 응답 경로: 추가 정보가 필요할 때 사용자에게 요청
그래프의 상태는 다음을 추적합니다.
- 대화 내역(사용자와 에이전트 간 메시지)
- 대화에서 추출된 모든 고객 및 구매 정보
- 사용자에게 보낼 다음 메시지 등
평가
최종 응답 평가자
에이전트의 최종 응답이 정답과 일치하는지 평가합니다.
from langsmith.evaluation import EvaluationClient
client = EvaluationClient()
async def final_answer_correct(outputs: dict, reference_outputs: dict, inputs: dict, **kwargs) -> bool:
# outputs, reference_outputs는 모두 {"response": ...} 형태의 딕셔너리여야 합니다.
user = f"""QUESTION: {inputs['question']}
GROUND TRUTH RESPONSE: {reference_outputs['response']}
STUDENT RESPONSE: {outputs['response']}"""
grade = await grader_llm.ainvoke([{"role": "system", "content": grader_instructions}, {"role": "user", "content": user}])
return grade["is_correct"]평가 실행 예시:
# Target function
async def run_graph(inputs: dict) -> dict:
result = await graph.ainvoke({"messages": [
{ "role": "user", "content": inputs['question']},
]}, config={"env": "test"})
return {"response": result["followup"]}
experiment_results = await client.aevaluate(
run_graph,
data=dataset_name,
evaluators=[final_answer_correct],
experiment_prefix="sql-agent-gpt4o-e2e",
num_repetitions=1,
max_concurrency=4,
)
experiment_results.to_pandas()경로(trajectory) 평가자
에이전트가 올바른 경로(예: 도구 호출 순서)를 따랐는지 평가합니다.
def trajectory_subsequence(outputs: dict, reference_outputs: dict) -> float:
"""에이전트가 원하는 단계 중 몇 개를 수행했는지 확인합니다."""
if len(reference_outputs['trajectory']) > len(outputs['trajectory']):
return False
i = j = 0
while i < len(reference_outputs['trajectory']) and j < len(outputs['trajectory']):
if reference_outputs['trajectory'][i] == outputs['trajectory'][j]:
i += 1
j += 1
return i / len(reference_outputs['trajectory'])평가 실행 예시:
async def run_graph(inputs: dict) -> dict:
trajectory = []
async for namespace, chunk in graph.astream({"messages": [
{
"role": "user",
"content": inputs['question'],
}
]}, subgraphs=True, stream_mode="debug"):
if chunk['type'] == 'task':
trajectory.append(chunk['payload']['name'])
if chunk['payload']['name'] == 'tools' and chunk['type'] == 'task':
for tc in chunk['payload']['input']['messages'][-1].tool_calls:
trajectory.append(tc['name'])
return {"trajectory": trajectory}
experiment_results = await client.aevaluate(
run_graph,
data=dataset_name,
evaluators=[trajectory_subsequence],
experiment_prefix="sql-agent-gpt4o-trajectory",
num_repetitions=1,
max_concurrency=4,
)
experiment_results.to_pandas()단일 단계 평가자
특정 단계(예: 의도 분류 등)를 개별적으로 평가할 수 있습니다.
# 데이터셋 생성
examples = [
{
"inputs": {"messages": [{"role": "user", "content": "i bought some tracks recently and i dont like them"}]},
"outputs": {"route": "refund_agent"},
},
{
"inputs": {"messages": [{"role": "user", "content": "I was thinking of purchasing some Rolling Stones tunes, any recommendations?"}]},
"outputs": {"route": "question_answering_agent"},
},
# ... (생략)
]
dataset_name = "Chinook Customer Service Bot: Intent Classifier"
if not client.has_dataset(dataset_name=dataset_name):
dataset = client.create_dataset(dataset_name=dataset_name)
client.create_examples(
dataset_id=dataset.id,
examples=examples
)
# 평가자
def correct(outputs: dict, reference_outputs: dict) -> bool:
return outputs["route"] == reference_outputs["route"]
# 평가 실행
async def run_intent_classifier(inputs: dict) -> dict:
command = await graph.nodes['intent_classifier'].ainvoke(inputs)
return {"route": command.goto}
experiment_results = await client.aevaluate(
run_intent_classifier,
data=dataset_name,
evaluators=[correct],
experiment_prefix="sql-agent-gpt4o-intent-classifier",
max_concurrency=4,
)참고 코드
전체 코드를 하나로 모은 예시는 원문을 참고하세요.
출처
반응형
