使用 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:
  """执行 Python 代码并返回结果。"""
  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="gpt-5.4-mini",
    tools=[code_tool, search_tool, polygon_aggregates],
    response_format=AgentOutputFormat,
    system_prompt="您是一位金融专家。请准确回答用户的问题",
)

创建一个 `tests/test_agent.py` 文件。

from app import agent, polygon_aggregates, search_tool # 从您的代理定义位置导入
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-5.4").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 评判者的指令
      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}"
      )
      # 运行评判 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