Skip to main content

Add evals to your app using Pytest or Vitest/Jest

This tutorial will show you how to use LangSmith's testing frameworks in both Python and JS/TS to write evals for your LLM application. In this tutorial, we will be creating and testing a ReAct agent that answers financial data questions related to publicly traded stocks. This tutorial uses LangGraph for agent orchestration, OpenAI's GPT-4o, Tavily for search, E2B's code interpreter and Polygon to retrieve stock data but it can be adapted for other frameworks, models and tools with minor modifications. Tavily, E2B and Polygon are free to sign up for.

Setup​

Installation​

First, install the packages required for making the agent:

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

Next, install the testing framework:

# Make sure you have langsmith>=0.3
pip install -U "langsmith[pytest]"

Environment Variables​

Set the following environment variables:

export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
export TAVILY_API_KEY=<YOUR_TAVILY_API_KEY>
export E2B_API_KEY=<YOUR_E2B_API_KEY>
export POLYGON_API_KEY=<YOUR_POLYGON_API_KEY>

Create your app​

To define our React agent, we will use LangGraph/LangGraph.js for the orchestation and LangChain for the LLM and tools.

Define tools​

First we are going to define the tools we are going to use in our agent. There are going to be 3 tools:

  • A search tool using Tavily
  • A code interpreter tool using E2B
  • A stock information tool using Polygon

Search Tool​

We will use Tavily for the search tool.

from langchain_community.tools import TavilySearchResults
search_tool = TavilySearchResults(
  max_results=5,
  include_raw_content=True,
)

Code Interpreter Tool​

We will use E2B for the code interpreter tool.

from e2b_code_interpreter import Sandbox
def code_tool(code: str) -> str:
  """Execute python code and return the result."""
  sbx = Sandbox()
  execution = sbx.run_code(code) 
  if execution.error:
      return f"Error: {execution.error}"
  return f"Results: {execution.results}, Logs: {execution.logs}"

Stock Information Tool​

We will use Polygon to get real-time data about stocks.

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

class TickerToolInput(TypedDict):
  """Input format for the ticker tool.
  
  The tool will pull data in aggregate blocks (timespan_multiplier * timespan) from the from_date to the to_date
  """
  ticker: Annotated[str, ..., "The ticker symbol of the stock"]
  timespan: Annotated[Literal["minute", "hour", "day", "week", "month", "quarter", "year"], ..., "The size of the time window."]
  timespan_multiplier: Annotated[int, ..., "The multiplier for the time window"]
  from_date: Annotated[str, ..., "The date to start pulling data from, YYYY-MM-DD format - ONLY include the year month and day"]
  to_date: Annotated[str, ..., "The date to stop pulling data, YYYY-MM-DD format - ONLY include the year month and day"]

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

def ticker_tool(query: TickerToolInput) -> str:
  """Pull data for the ticker."""
  return polygon_aggregate.invoke(query)

Define agent​

Now that we have defined all of our tools, we can use LangGraph's create_react_agent/createReactAgent to create our agent.

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from typing_extensions import Annotated, TypedDict, Optional
    
class AgentOutputFormat(TypedDict):
  numeric_answer: Annotated[Optional[float], ..., "The numeric answer, if the user asked for one"]
  text_answer: Annotated[Optional[str], ..., "The text answer, if the user asked for one"]
  reasoning: Annotated[str, ..., "The reasoning behind the answer"]

tools = [code_tool, search_tool, ticker_tool]

agent = create_react_agent(
  model=ChatOpenAI(model="gpt-4o-mini"),
  tools=tools,
  response_format=AgentOutputFormat,
  state_modifier="You are a financial expert. Respond to the users query accurately",
)

Write tests​

Now that we have defined our agent, let's write a few tests to ensure basic functionality. In this tutorial we are going to test whether the agent's tool calling abilities are working, whether the agent knows to ignore irrelevant questions, and whether it is able to answer complex questions that involve using all of the tools.

Setup test file​

We need to first set up a test file and add the imports needed at the top of the file.

Name your test file test_evals.py

from agent import agent # import from wherever your agent is defined
import pytest
import json
from langsmith import testing

Test 1: Ignore irrelevant questions​

The first test will be a simple check that the agent does not use tools on irrelevant queries.

@pytest.mark.langsmith
@pytest.mark.parametrize( # Can still use all normal pytest markers
  "query",
  [
      "Hello!",
      "How are you doing?"
  ],
)
def test_no_tools_on_unrelated_query(query):
  testing.log_inputs({"query": query})
  # Test that the agent does not use tools on unrelated queries
  result = agent.invoke({"messages": [("user", query)]})
  testing.log_outputs({"result": result})
  
  testing.log_reference_outputs({"numMessages": 2})
  
  # Check that the flow was HUMAN -> AI FINAL RESPONSE (no tools called)
  assert len(result['messages']) == 2

Test 2: Tool calling​

For tool calling, we are going to verify that the agent calls the correct tool with the correct parameters.

@pytest.mark.langsmith
def test_searches_for_correct_ticker():
  user_query = "What is the price of Apple?"
  testing.log_inputs({"user_query": user_query})
  # Test that the agent searches for the correct ticker
  result = agent.invoke({"messages": [("user", user_query)]})
  testing.log_outputs({"result": result})
  
  testing.log_reference_outputs({"ticker": "AAPL", "numMessages": 4})
  
  # The agent should have made a single tool call to the ticker tool
  ticker = json.loads(result['messages'][1].additional_kwargs['tool_calls'][0]['function']['arguments'])["query"]["ticker"]
  
  # Check that the right ticker was queried
  assert ticker == "AAPL"
  
  # Check that the flow was HUMAN -> AI -> TOOL -> AI FINAL RESPONSE
  assert len(result['messages']) == 4

Test 3: Complex question answering​

For answering a complex question, we are going to ensure that the agent uses the coding tool for any calculations, and also log the total number of steps taken so we can track over time how the performance of the agent improves.

@pytest.mark.langsmith
def test_executes_code_when_needed():
  user_query = "What was the average return rate for FAANG stock in 2024?"
  testing.log_inputs({"user_query": user_query})
  
  # Test that the agent executes code when needed
  result = agent.invoke({"messages": [("user", user_query)]})
  testing.log_outputs({"result": result})
  
  testing.log_reference_outputs({"answer": 53})
  
  # Grab all the tool calls made by the LLM
  tool_calls = [tc['function']['name'] for m in result['messages'] if m.additional_kwargs and 'tool_calls' in m.additional_kwargs for tc in m.additional_kwargs['tool_calls'] ]
  
  # This will log the number of steps taken by the LLM, which we can track over time to measure performance
  testing.log_feedback(key="num_steps", value=len(result['messages']) - 1)
  
  # Assert that the code tool was used
  assert "code_tool" in tool_calls
  
  # Assert that the answer is within 1 of the expected value
  assert abs(result['structured_response']['numeric_answer'] - 53) <= 1

Test 4: LLM-as-a-judge​

We are going to ensure that the agent's answer is grounded in the search results by running an LLM-as-a-judge evaluation. In order to trace the LLM as a judge call separately from our agent, we will use the LangSmith provided trace_feedback context manager in Python and wrapEvaluator function in JS/TS.

judge_llm = ChatOpenAI(model="gpt-4o")

@pytest.mark.langsmith
def test_grounded_in_the_source():
  # Test that the agent is grounded in the source
  user_query = "How did Nvidia stock do in 2024 according to analysts?"
  t.log_inputs({"user_query": user_query})
  
  result = agent.invoke({"messages": [("user", user_query)]})    
  t.log_outputs({"result": result})

  # Grab all the search calls made by the LLM
  tavily_tool_message_indices = []
  for i, m in enumerate(result['messages']):
      if m.additional_kwargs and 'tool_calls' in m.additional_kwargs and 
      m.additional_kwargs['tool_calls'][0]['function']['name'] == 'tavily_search_results_json':
          tavily_tool_message_indices.append(i+1)

  documents = "\n".join([m.content for i,m in enumerate(result['messages']) if i in tavily_tool_message_indices])

  with testing.trace_feedback():
      # Instructions for the LLM judge
      instructions = (
          "Return 1 if the ANSWER is grounded in the DOCUMENTS"
          "Return 0 if the ANSWER is not grounded in the DOCUMENTS"
      )

      # Run the judge LLM
      grade = judge_llm.invoke([
          {"role": "system", "content": instructions},
          {"role": "user", "content": f"ANSWER: {result['structured_response']['text_answer']}\nDOCUMENTS: {documents}"},
      ])

      score = int(grade.content)
      testing.log_feedback(key="groundedness", score=score)
  assert score

Run tests​

Once you have setup your config files (if you are using Vitest or Jest), you can run your tests using the following commands:

Config files for Vitest/Jest
ls.vitest.config.ts
import { defineConfig } from "vitest/config";

export default defineConfig({
  test: { 
    include: ["**/*.eval.?(c|m)[jt]s"],
    reporters: ["langsmith/vitest/reporter"],
    setupFiles: ["dotenv/config"],
  },
});
pytest --output='ls' tests

Reference code​

Remember to also add the config files for Vitest and Jest to your project.

Agent​

Agent code
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
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

def code_tool(code: str) -> str:
  """Execute python code and return the result."""
  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):
  """Input format for the ticker tool.
  
  The tool will pull data in aggregate blocks (timespan_multiplier * timespan) from the from_date to the to_date
  """
  ticker: Annotated[str, ..., "The ticker symbol of the stock"]
  timespan: Annotated[Literal["minute", "hour", "day", "week", "month", "quarter", "year"], ..., "The size of the time window."]
  timespan_multiplier: Annotated[int, ..., "The multiplier for the time window"]
  from_date: Annotated[str, ..., "The date to start pulling data from, YYYY-MM-DD format - ONLY include the year month and day"]
  to_date: Annotated[str, ..., "The date to stop pulling data, YYYY-MM-DD format - ONLY include the year month and day"]

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

def ticker_tool(query: TickerToolInput) -> str:
  """Pull data for the ticker."""
  return polygon_aggregate.invoke(query)

search_tool = TavilySearchResults(
  max_results=5,
  include_raw_content=True,
)

class AgentOutputFormat(TypedDict):
  numeric_answer: Annotated[Optional[float], ..., "The numeric answer, if the user asked for one"]
  text_answer: Annotated[Optional[str], ..., "The text answer, if the user asked for one"]
  reasoning: Annotated[str, ..., "The reasoning behind the answer"]

tools = [code_tool, search_tool, ticker_tool]

agent = create_react_agent(
  model=ChatOpenAI(model="gpt-4o-mini"),
  tools=tools,
  response_format=AgentOutputFormat,
  state_modifier="You are a financial expert. Respond to the users query accurately",
)

Tests​

Test code
from agent import agent
import pytest
import json
from langsmith import testing
from langchain_openai import ChatOpenAI

@pytest.mark.langsmith
@pytest.mark.parametrize(
  "query",
  [
      "Hello!",
      "How are you doing?"
  ],
)
def test_no_tools_on_unrelated_query(query):
  # Test that the agent does not use tools on unrelated queries
  result = agent.invoke({"messages": [("user", query)]})
  # first human message and then the AI response
  assert len(result['messages']) == 2

@pytest.mark.langsmith
def test_searches_for_correct_ticker():
  # Test that the agent searches for the correct ticker
  result = agent.invoke({"messages": [("user", "What is the price of Aple?")]})
  ticker = json.loads(result['messages'][1].additional_kwargs['tool_calls'][0]['function']['arguments'])["query"]["ticker"]
  assert ticker == "AAPL"
  assert len(result['messages']) == 4


@pytest.mark.langsmith
def test_executes_code_when_needed():
  # Test that the agent executes code when needed
  result = agent.invoke({"messages": [("user", "What was the average return rate for FAANG stock in 2024?")]})
  tool_calls = [tc['function']['name'] for m in result['messages'] if m.additional_kwargs and 'tool_calls' in m.additional_kwargs for tc in m.additional_kwargs['tool_calls'] ]
  testing.log_feedback(key="num_steps", score=len(result['messages']) - 1)
  assert "code_tool" in tool_calls
  assert abs(result['structured_response']['numeric_answer'] - 53) <= 1

judge_llm = ChatOpenAI(model="gpt-4o")

@pytest.mark.langsmith
def test_grounded_in_the_source():
  # Test that the agent is grounded in the source
  result = agent.invoke({"messages": [("user", "How did Nvidia stock do in 2024 according to analysts?")]})    

  tavily_tool_message_indices = []
  for i, m in enumerate(result['messages']):
      if m.additional_kwargs and 'tool_calls' in m.additional_kwargs and m.additional_kwargs['tool_calls'][0]['function']['name'] == 'tavily_search_results_json':
          tavily_tool_message_indices.append(i+1)

  documents = "\n".join([m.content for i,m in enumerate(result['messages']) if i in tavily_tool_message_indices])

  with testing.trace_feedback():
      instructions = (
          "Return 1 if the ANSWER is grounded in the DOCUMENTS"
          "Return 0 if the ANSWER is not grounded in the DOCUMENTS"
      )

      grade = judge_llm.invoke([
          {"role": "system", "content": instructions},
          {"role": "user", "content": f"ANSWER: {result['structured_response']['text_answer']}\nDOCUMENTS: {documents}"},
      ])

      score = int(grade.content)
      testing.log_feedback(key="groundedness", score=score)
  assert score

Was this page helpful?


You can leave detailed feedback on GitHub.