그래프 평가하는 방법
그래프 평가하는 방법
langgraph는 LLM으로 상태 기반의 다중 에이전트 애플리케이션을 구축하기 위한 라이브러리로, 에이전트 및 멀티 에이전트 워크플로우를 만드는 데 사용돼요. langgraph 그래프를 평가하는 것은 단일 호출에 많은 LLM 호출이 포함될 수 있고, 어떤 LLM 호출이 이루어질지는 이전 호출의 출력에 따라 달라질 수 있기 때문에 까다로울 수 있습니다. 이 가이드에서는 그래프와 그래프 노드를 evaluate() / aevaluate()에 전달하는 방법의 메커니즘에 초점을 맞출게요. 에이전트를 구축할 때의 평가 기법과 모범 사례에 대해서는 LangSmith 평가 문서를 참고하세요.
출처: 문서
본문
엔드투엔드 평가 (End-to-end evaluations)
가장 일반적인 평가 유형은 엔드투엔드 평가로, 각 예시 입력에 대한 최종 그래프 출력을 평가하는 방식입니다.
그래프 정의하기 (Define a graph)
시작하기 위해 간단한 ReACT 에이전트를 만들어 보겠습니다:
from typing import Annotated, Literal, TypedDict
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
# Messages have the type "list". The 'add_messages' function
# in the annotation defines how this state key should be updated
# (in this case, it appends messages to the list, rather than overwriting them)
messages: Annotated[list, add_messages]
# Define the tools for the agent to use
@tool
def search(query: str) -> str:
"""Call to surf the web."""
# This is a placeholder, but don't tell the LLM that...
if "sf" in query.lower() or "san francisco" in query.lower():
return "It's 60 degrees and foggy."
return "It's 90 degrees and sunny."
tools = [search]
tool_node = ToolNode(tools)
model = init_chat_model("claude-sonnet-4-6").bind_tools(tools)
# Define the function that determines whether to continue or not
def should_continue(state: State) -> Literal["tools", END]:
messages = state['messages']
last_message = messages[-1]
# If the LLM makes a tool call, then we route to the "tools" node
if last_message.tool_calls:
return "tools"
# Otherwise, we stop (reply to the user)
return END
# Define the function that calls the model
def call_model(state: State):
messages = state['messages']
response = model.invoke(messages)
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define a new graph
workflow = StateGraph(State)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
# Set the entrypoint as 'agent'
# This means that this node is the first one called
workflow.add_edge(START, "agent")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use 'agent'.
# This means these are the edges taken after the 'agent' node is called.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
)
# We now add a normal edge from 'tools' to 'agent'.
# This means that after 'tools' is called, 'agent' node is called next.
workflow.add_edge("tools", 'agent')
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable.
# Note that we're (optionally) passing the memory when compiling the graph
app = workflow.compile()
데이터셋 생성하기 (Create a dataset)
질문과 기대 응답으로 구성된 간단한 데이터셋을 만들어 보겠습니다:
from langsmith import Client
questions = [
"what's the weather in sf",
"what's the weather in san fran",
"what's the weather in tangier"
]
answers = [
"It's 60 degrees and foggy.",
"It's 60 degrees and foggy.",
"It's 90 degrees and sunny.",
]
ls_client = Client()
dataset = ls_client.create_dataset("weather agent")
ls_client.create_examples(
inputs=[{"question": q} for q in questions],
outputs=[{"answer": a} for a in answers],
dataset_id=dataset.id,
)
평가기 생성하기 (Create an evaluator)
그리고 간단한 평가기:
langsmith>=0.2.0 필요
judge_llm = init_chat_model("gpt-5.5")
async def correct(outputs: dict, reference_outputs: dict) -> bool:
instructions = (
"Given an actual answer and an expected answer, determine whether"
" the actual answer contains all of the information in the"
" expected answer. Respond with 'CORRECT' if the actual answer"
" does contain all of the expected information and 'INCORRECT'"
" otherwise. Do not include anything else in your response."
)
# Our graph outputs a State dictionary, which in this case means
# we'll have a 'messages' key and the final message should
# be our actual answer.
actual_answer = outputs["messages"][-1].content
expected_answer = reference_outputs["answer"]
user_msg = (
f"ACTUAL ANSWER: {actual_answer}"
f"\n\nEXPECTED ANSWER: {expected_answer}"
)
response = await judge_llm.ainvoke(
[
{"role": "system", "content": instructions},
{"role": "user", "content": user_msg}
]
)
return response.content.upper() == "CORRECT"
평가 실행하기 (Run evaluations)
이제 평가를 실행하고 결과를 살펴볼 수 있어요. 그래프 함수를 예시에 저장된 형식의 입력을 받을 수 있도록 래핑하기만 하면 됩니다:
그래프의 모든 노드가 동기 함수로 정의되어 있다면
evaluate또는aevaluate를 사용할 수 있어요. 노드 중 하나라도 비동기로 정의되어 있다면aevaluate를 사용해야 합니다.
langsmith>=0.2.0 필요
import asyncio
from langsmith import aevaluate
def example_to_state(inputs: dict) -> dict:
return {"messages": [{"role": "user", "content": inputs['question']}]}
# We use LCEL declarative syntax here.
# Remember that langgraph graphs are also langchain runnables.
target = example_to_state | app
async def main():
experiment_results = await aevaluate(
target,
data="weather agent",
evaluators=[correct],
max_concurrency=4, # optional
experiment_prefix="claude-sonnet-4-6-baseline", # optional
metadata={ # optional, used to populate model/prompt/tool columns in UI
"models": "google_genai:gemini-3.6-flash",
"tools": [{"name": "search", "description": "Call to surf the web."}],
},
)
print(experiment_results)
asyncio.run(main())
중간 단계 평가하기 (Evaluating intermediate steps)
에이전트의 최종 출력뿐만 아니라 에이전트가 수행한 중간 단계도 평가하는 것이 유용할 때가 많아요. langgraph의 좋은 점은 그래프의 출력이 상태 객체이며, 종종 이미 수행된 중간 단계에 대한 정보를 담고 있다는 것입니다. 보통 우리는 상태의 메시지만 살펴봐도 관심 있는 것을 평가할 수 있습니다. 예를 들어 메시지를 살펴보고 모델이 첫 단계로 'search' 도구를 호출했는지 확인할 수 있습니다.
langsmith>=0.2.0 필요
def right_tool(outputs: dict) -> bool:
tool_calls = outputs["messages"][1].tool_calls
return bool(tool_calls and tool_calls[0]["name"] == "search")
async def main():
experiment_results = await aevaluate(
target,
data="weather agent",
evaluators=[correct, right_tool],
max_concurrency=4, # optional
experiment_prefix="claude-sonnet-4-6-baseline", # optional
metadata={ # optional, used to populate model/prompt/tool columns in UI
"models": "google_genai:gemini-3.6-flash",
"tools": [{"name": "search", "description": "Call to surf the web."}],
},
)
print(experiment_results)
상태에 없는 중간 단계에 대한 정보가 필요하다면 Run 객체를 살펴볼 수 있어요. 여기에는 모든 노드 입력과 출력에 대한 전체 트레이스가 포함되어 있습니다:
커스텀 평가기에 전달할 수 있는 인자에 대한 자세한 내용은 이 하우투 가이드에서 확인하세요.
from langsmith.schemas import Run, Example
def right_tool_from_run(run: Run, example: Example) -> dict:
# Get documents and answer
first_model_run = next(run for run in root_run.child_runs if run.name == "agent")
tool_calls = first_model_run.outputs["messages"][-1].tool_calls
right_tool = bool(tool_calls and tool_calls[0]["name"] == "search")
return {"key": "right_tool", "value": right_tool}
async def main():
experiment_results = await aevaluate(
target,
data="weather agent",
evaluators=[correct, right_tool_from_run],
max_concurrency=4, # optional
experiment_prefix="claude-sonnet-4-6-baseline", # optional
metadata={ # optional, used to populate model/prompt/tool columns in UI
"models": "google_genai:gemini-3.6-flash",
"tools": [{"name": "search", "description": "Call to surf the web."}],
},
)
print(experiment_results)
개별 노드 실행 및 평가하기 (Running and evaluating individual nodes)
때로는 시간과 비용을 절약하기 위해 단일 노드를 직접 평가하고 싶을 때가 있어요. langgraph는 이를 쉽게 해줍니다. 이 경우에도 계속 사용해 온 평가기를 그대로 사용할 수 있습니다.
node_target = example_to_state | app.nodes["agent"]
async def main():
node_experiment_results = await aevaluate(
node_target,
data="weather agent",
evaluators=[right_tool_from_run],
max_concurrency=4, # optional
experiment_prefix="claude-sonnet-4-6-model-node", # optional
metadata={ # optional, used to populate model/prompt/tool columns in UI
"models": "google_genai:gemini-3.6-flash",
"tools": [{"name": "search", "description": "Call to surf the web."}],
},
)
print(node_experiment_results)
참고 코드 (Reference code)
Define a graph
class State(TypedDict): # Messages have the type "list". The 'add_messages' function # in the annotation defines how this state key should be updated # (in this case, it appends messages to the list, rather than overwriting them) messages: Annotated[list, add_messages]
Define the tools for the agent to use
@tool def search(query: str) -> str: """Call to surf the web.""" # This is a placeholder, but don't tell the LLM that... if "sf" in query.lower() or "san francisco" in query.lower(): return "It's 60 degrees and foggy." return "It's 90 degrees and sunny."
tools = [search] tool_node = ToolNode(tools) model = init_chat_model("claude-sonnet-4-6").bind_tools(tools)
Define the function that determines whether to continue or not
def should_continue(state: State) -> Literal["tools", END]: messages = state['messages'] last_message = messages[-1]
# If the LLM makes a tool call, then we route to the "tools" node
if last_message.tool_calls:
return "tools"
# Otherwise, we stop (reply to the user)
return END
Define the function that calls the model
def call_model(state: State): messages = state['messages'] response = model.invoke(messages) # We return a list, because this will get added to the existing list return {"messages": [response]}
Define a new graph
workflow = StateGraph(State)
Define the two nodes we will cycle between
workflow.add_node("agent", call_model) workflow.add_node("tools", tool_node)
Set the entrypoint as 'agent'
This means that this node is the first one called
workflow.add_edge(START, "agent")
We now add a conditional edge
workflow.add_conditional_edges( # First, we define the start node. We use 'agent'. # This means these are the edges taken after the 'agent' node is called. "agent", # Next, we pass in the function that will determine which node is called next. should_continue, )
We now add a normal edge from 'tools' to 'agent'.
This means that after 'tools' is called, 'agent' node is called next.
workflow.add_edge("tools", 'agent')
Finally, we compile it!
This compiles it into a LangChain Runnable,
meaning you can use it as you would any other runnable.
Note that we're (optionally) passing the memory when compiling the graph
app = workflow.compile()
questions = [ "what's the weather in sf", "what's the weather in san fran", "what's the weather in tangier" ]
answers = [ "It's 60 degrees and foggy.", "It's 60 degrees and foggy.", "It's 90 degrees and sunny.", ]
Create a dataset
ls_client = Client() dataset = ls_client.create_dataset("weather agent") ls_client.create_examples( inputs=[{"question": q} for q in questions], outputs=[{"answer": a} for a in answers], dataset_id=dataset.id, )
Define evaluators
judge_llm = init_chat_model("gpt-5.5")
async def correct(outputs: dict, reference_outputs: dict) -> bool: instructions = ( "Given an actual answer and an expected answer, determine whether" " the actual answer contains all of the information in the" " expected answer. Respond with 'CORRECT' if the actual answer" " does contain all of the expected information and 'INCORRECT'" " otherwise. Do not include anything else in your response." ) # Our graph outputs a State dictionary, which in this case means # we'll have a 'messages' key and the final message should # be our actual answer. actual_answer = outputs["messages"][-1].content expected_answer = reference_outputs["answer"] user_msg = ( f"ACTUAL ANSWER: {actual_answer}" f"\n\nEXPECTED ANSWER: {expected_answer}" ) response = await judge_llm.ainvoke( [ {"role": "system", "content": instructions}, {"role": "user", "content": user_msg} ] ) return response.content.upper() == "CORRECT"
def right_tool(outputs: dict) -> bool: tool_calls = outputs["messages"][1].tool_calls return bool(tool_calls and tool_calls[0]["name"] == "search")
def example_to_state(inputs: dict) -> dict: return {"messages": [{"role": "user", "content": inputs['question']}]}
We use LCEL declarative syntax here.
Remember that langgraph graphs are also langchain runnables.
target = example_to_state | app
Run evaluation
async def main(): experiment_results = await aevaluate( target, data="weather agent", evaluators=[correct, right_tool], max_concurrency=4, # optional experiment_prefix="claude-sonnet-4-6-baseline", # optional metadata={ # optional, used to populate model/prompt/tool columns in UI "models": "google_genai:gemini-3.6-flash", "tools": [{"name": "search", "description": "Call to surf the web."}], }, ) print(experiment_results)
asyncio.run(main())
</Accordion>
## 더 알아보기 (Learn more)
- [`langgraph` 평가 문서](/langsmith/evaluation)