Pytest/Vitest와 LangSmith로 ReAct 에이전트 테스트하기

pip install -U langgraph langchain[openai] langchain-community e2b-code-interpreter

# langsmith>=0.3.1 버전이 설치되어 있는지 확인하세요
pip install -U "langsmith[pytest]"

from langchain_community.tools import TavilySearchResults
from e2b_code_interpreter import Sandbox
from langchain_community.tools.polygon.aggregates import PolygonAggregates
from langchain_community.utilities.polygon import PolygonAPIWrapper
from typing_extensions import Annotated, TypedDict, Optional, Literal

# 검색 도구 정의
search_tool = TavilySearchResults(
  max_results=5,
  include_raw_content=True,
)

# 코드 도구 정의
def code_tool(code: str) -> str:
  """파이썬 코드를 실행하고 결과를 반환합니다."""
  sbx = Sandbox()
  execution = sbx.run_code(code)

  if execution.error:
      return f"Error: {execution.error}"
  return f"Results: {execution.results}, Logs: {execution.logs}"

# 주식 티커 도구의 입력 스키마 정의
class TickerToolInput(TypedDict):
  """티커 도구의 입력 형식입니다.
    이 도구는 from_date부터 to_date까지 집계 블록(timespan_multiplier * timespan)으로 데이터를 가져옵니다.
  """
  ticker: Annotated[str, ..., "주식의 티커 심볼"]
  timespan: Annotated[Literal["minute", "hour", "day", "week", "month", "quarter", "year"], ..., "시간 윈도우의 크기"]
  timespan_multiplier: Annotated[int, ..., "시간 윈도우의 배수"]
  from_date: Annotated[str, ..., "데이터를 가져올 시작 날짜, YYYY-MM-DD 형식 - 년, 월, 일만 포함"]
  to_date: Annotated[str, ..., "데이터를 가져올 종료 날짜, YYYY-MM-DD 형식 - 년, 월, 일만 포함"]

api_wrapper = PolygonAPIWrapper()
polygon_aggregate = PolygonAggregates(api_wrapper=api_wrapper)

# 주식 티커 도구 정의
def ticker_tool(query: TickerToolInput) -> str:
  """티커 데이터를 가져옵니다."""
  return polygon_aggregate.invoke(query)

from typing_extensions import Annotated, TypedDict
from langchain.agents import create_agent


class AgentOutputFormat(TypedDict):
    numeric_answer: Annotated[float | None, ..., "사용자가 숫자 답변을 요청한 경우 숫자 답변"]
    text_answer: Annotated[str | None, ..., "사용자가 텍스트 답변을 요청한 경우 텍스트 답변"]
    reasoning: Annotated[str, ..., "답변의 근거"]

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[code_tool, search_tool, polygon_aggregates],
    response_format=AgentOutputFormat,
    system_prompt="You are a financial expert. Respond to the users query accurately",
)

`tests/test_agent.py` 파일을 생성합니다.

from app import agent, polygon_aggregates, search_tool # 에이전트가 정의된 곳에서 import합니다
import pytest
from langsmith import testing as t

@pytest.mark.langsmith
@pytest.mark.parametrize(
  # <-- 모든 일반 pytest 마커를 계속 사용할 수 있습니다
  "query",
  ["Hello!", "How are you doing?"],
)
def test_no_tools_on_offtopic_query(query: str) -> None:
  """에이전트가 주제에서 벗어난 쿼리에 대해 도구를 사용하지 않는지 테스트합니다."""
  # 테스트 예제를 기록합니다
  t.log_inputs({"query": query})
  expected = []
  t.log_reference_outputs({"tool_calls": expected})
  # ReACT 루프를 실행하는 대신 에이전트의 모델 노드를 직접 호출합니다.
  result = agent.nodes["agent"].invoke(
      {"messages": [{"role": "user", "content": query}]}
  )
  actual = result["messages"][0].tool_calls
  t.log_outputs({"tool_calls": actual})
  # 도구 호출이 없었는지 확인합니다.
  assert actual == expected

@pytest.mark.langsmith
def test_searches_for_correct_ticker() -> None:
  """모델이 간단한 쿼리에 대해 올바른 티커를 조회하는지 테스트합니다."""
  # 테스트 예제를 기록합니다
  query = "What is the price of Apple?"
  t.log_inputs({"query": query})
  expected = "AAPL"
  t.log_reference_outputs({"ticker": expected})
  # 전체 ReACT 루프를 실행하는 대신 에이전트의 모델 노드를 직접 호출합니다.
  result = agent.nodes["agent"].invoke(
      {"messages": [{"role": "user", "content": query}]}
  )
  tool_calls = result["messages"][0].tool_calls
  if tool_calls[0]["name"] == polygon_aggregates.name:
      actual = tool_calls[0]["args"]["ticker"]
  else:
      actual = None
  t.log_outputs({"ticker": actual})
  # 올바른 티커가 쿼리되었는지 확인합니다
  assert actual == expected

@pytest.mark.langsmith
def test_executes_code_when_needed() -> None:
  query = (
      "In the past year Facebook stock went up by 66.76%, "
      "Apple by 25.24%, Google by 37.11%, Amazon by 47.52%, "
      "Netflix by 78.31%. Whats the avg return in the past "
      "year of the FAANG stocks, expressed as a percentage?"
  )
  t.log_inputs({"query": query})
  expected = 50.988
  t.log_reference_outputs({"response": expected})
  # 필요할 때 에이전트가 코드를 실행하는지 테스트합니다
  result = agent.invoke({"messages": [{"role": "user", "content": query}]})
  t.log_outputs({"result": result["structured_response"].get("numeric_answer")})
  # LLM이 수행한 모든 도구 호출을 가져옵니다
  tool_calls = [
      tc["name"]
      for msg in result["messages"]
      for tc in getattr(msg, "tool_calls", [])
  ]
  # 에이전트가 수행한 단계 수를 기록합니다. 이는 에이전트가 답변에 도달하는
  # 효율성을 판단하는 데 유용합니다.
  t.log_feedback(key="num_steps", score=len(result["messages"]) - 1)
  # 코드 도구가 사용되었는지 확인합니다
  assert "code_tool" in tool_calls
  # 숫자 답변이 제공되었는지 확인합니다:
  assert result["structured_response"].get("numeric_answer") is not None
  # 답변이 정확한지 확인합니다
  assert abs(result["structured_response"]["numeric_answer"] - expected) <= 0.01

from typing_extensions import Annotated, TypedDict
from langchain.chat_models import init_chat_model

class Grade(TypedDict):
  """소스 문서에서 답변의 근거를 평가합니다."""
  score: Annotated[
      bool,
      ...,
      "답변이 소스 문서에 완전히 근거한 경우 True를, 그렇지 않으면 False를 반환합니다.",
  ]

judge_llm = init_chat_model("gpt-4o").with_structured_output(Grade)

@pytest.mark.langsmith
def test_grounded_in_source_info() -> None:
  """응답이 도구 출력에 근거하고 있는지 테스트합니다."""
  query = "How did Nvidia stock do in 2024 according to analysts?"
  t.log_inputs({"query": query})
  result = agent.invoke({"messages": [{"role": "user", "content": query}]})
  # LLM이 수행한 모든 검색 호출을 가져옵니다
  search_results = "\n\n".join(
      msg.content
      for msg in result["messages"]
      if msg.type == "tool" and msg.name == search_tool.name
  )
  t.log_outputs(
      {
          "response": result["structured_response"].get("text_answer"),
          "search_results": search_results,
      }
  )
  # 피드백 LLM 실행을 에이전트 실행과 별도로 추적합니다.
  with t.trace_feedback():
      # LLM judge를 위한 지침
      instructions = (
          "Grade the following ANSWER. "
          "The ANSWER should be fully grounded in (i.e. supported by) the source DOCUMENTS. "
          "Return True if the ANSWER is fully grounded in the DOCUMENTS. "
          "Return False if the ANSWER is not grounded in the DOCUMENTS."
      )
      answer_and_docs = (
          f"ANSWER: {result['structured_response'].get('text_answer', '')}\n"
          f"DOCUMENTS:\n{search_results}"
      )
      # judge LLM을 실행합니다
      grade = judge_llm.invoke(
          [
              {"role": "system", "content": instructions},
              {"role": "user", "content": answer_and_docs},
          ]
      )
      t.log_feedback(key="groundedness", score=grade["score"])
  assert grade['score']

pytest --langsmith-output tests