使用 Pytest/Vitest 和 LangSmith 测试 ReAct 代理

本教程将向您展示如何使用 LangSmith 与流行测试工具（Pytest、Vitest 和 Jest）的集成来评估您的 LLM 应用程序。我们将创建一个回答公开交易股票问题的 ReAct 代理，并为其编写全面的测试套件。

设置

本教程使用 LangGraph 进行代理编排，OpenAI 的 GPT-4o 作为 LLM，Tavily 进行搜索，E2B 的代码解释器，以及 Polygon 获取股票数据，但通过少量修改即可适用于其他框架、模型和工具。Tavily、E2B 和 Polygon 均可免费注册。

安装

首先，安装创建代理所需的包：

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

yarn add @langchain/openai @langchain/community @langchain/langgraph @langchain/core @e2b/code-interpreter @polygon.io/client-js openai zod

接下来，安装测试框架：

# 确保您拥有 langsmith>=0.3.1
pip install -U "langsmith[pytest]"

yarn add -D langsmith vitest

yarn add -D langsmith jest

环境变量

设置以下环境变量：

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=<YOUR_LANGSMITH_API_KEY>
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>

创建您的应用

为了定义我们的 ReAct 代理，我们将使用 LangGraph/LangGraph.js 进行编排，使用 LangChain 处理 LLM 和工具。

定义工具

首先，我们将定义代理中要使用的工具。将有 3 个工具：

使用 Tavily 的搜索工具
使用 E2B 的代码解释器工具
使用 Polygon 的股票信息工具

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)

import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { Sandbox } from "@e2b/code-interpreter";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { restClient } from "@polygon.io/client-js";

// 定义搜索工具
const searchTool = new TavilySearchResults({
  maxResults: 5,
});

// 定义代码工具
const codeTool = tool(async (input) => {
  const sbx = await Sandbox.create();
  const execution = await sbx.runCode(input.code);
  if (execution.error) {
    return `Error: ${execution.error}`;
  }
  return `Results: ${execution.results}, Logs: ${execution.logs}`;
}, {
  name: "code",
  description: "执行 Python 代码并返回结果。",
  schema: z.object({
    code: z.string().describe("要执行的 Python 代码"),
  }),
});

// 定义股票代码工具的输入模式
const TickerToolInputSchema = z.object({
  ticker: z.string().describe("股票的代码符号"),
  timespan: z.enum(["minute", "hour", "day", "week", "month", "quarter", "year"]).describe("时间窗口的大小。"),
  timespan_multiplier: z.number().describe("时间窗口的乘数"),
  from_date: z
    .string()
    .describe("开始拉取数据的日期，YYYY-MM-DD 格式 - 仅包含年月日"),
  to_date: z
    .string()
    .describe("停止拉取数据的日期，YYYY-MM-DD 格式 - 仅包含年月日"),
});

const rest = restClient(process.env.POLYGON_API_KEY);

// 定义股票代码工具
const tickerTool = tool(async (query) => {
  const parsed = TickerToolInputSchema.parse(query);
  const result = await rest.stocks.aggregates(
      parsed.ticker,
      parsed.timespan_multiplier,
      parsed.timespan,
      parsed.from_date,
      parsed.to_date
  );
  return JSON.stringify(result);
}, {
  name: "ticker",
  description: "拉取股票代码的数据",
  schema: TickerToolInputSchema,
});

定义代理

现在我们已经定义了所有工具，可以使用 create_agent 来创建我们的代理。

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="您是一位金融专家。请准确回答用户的问题",
)

import { z } from "zod";
import { ChatOpenAI } from "@langchain/openai";
import { createReactAgent } from "@langchain/langgraph/prebuilt";

const AgentOutputFormatSchema = z.object({
  numeric_answer: z.number().optional().describe("数值答案，如果用户要求提供"),
  text_answer: z.string().optional().describe("文本答案，如果用户要求提供"),
  reasoning: z.string().describe("答案背后的推理过程"),
})

const tools = [codeTool, searchTool, tickerTool];

const agent = createReactAgent({
  llm: new ChatOpenAI({ model: "gpt-5.4" }),
  tools: tools,
  responseFormat: AgentOutputFormatSchema,
  stateModifier: "您是一位金融专家。请准确回答用户的问题",
});

export default agent;

编写测试

现在我们已经定义了代理，让我们编写一些测试以确保基本功能。在本教程中，我们将测试代理的工具调用能力是否正常工作，代理是否知道忽略无关问题，以及它是否能够回答涉及使用所有工具的复杂问题。我们需要首先设置一个测试文件，并在文件顶部添加所需的导入。

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

from app import agent, polygon_aggregates, search_tool # 从您的代理定义位置导入
import pytest
from langsmith import testing as t

将您的测试文件命名为 `agent.vitest.eval.ts`

import { expect } from "vitest";
import * as ls from "langsmith/vitest";
import agent from "../agent"; // 从您的代理定义位置导入

// 可选，但建议将测试分组
ls.describe("Agent Tests", () => {
  // 在此处放置测试
});

将您的测试文件命名为 `agent.jest.eval.ts`

import { expect } from "@jest/globals";
import * as ls from "langsmith/jest";
import agent from "../agent"; // 从您的代理定义位置导入

// 可选，但建议将测试分组
ls.describe("Agent Tests", () => {
  // 在此处放置测试
});

测试 1：处理无关问题

第一个测试将简单检查代理是否不会在无关查询上使用工具。

@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

ls.test.each([
  { inputs: { query: "Hello!" }, expected: { numMessages: 2 } },
  { inputs: { query: "How are you doing?" }, expected: { numMessages: 2 } },
])(
  "should not use tools on offtopic query: %s",
  async ({ inputs: { query }, expected: { numMessages } }) => {
    const result = await agent.invoke({ messages: [{ role: "user", content: query }] });
    ls.logOutputs(result);

    // 检查流程是否为 HUMAN -> AI FINAL RESPONSE（未调用工具）
    expect(result.messages).toHaveLength(numMessages);
  }
);

ls.test.each([
  { inputs: { query: "Hello!" }, expected: { numMessages: 2 } },
  { inputs: { query: "How are you doing?" }, expected: { numMessages: 2 } },
])(
  "should not use tools on offtopic query: %s",
  async ({ inputs: { query }, expected: { numMessages } }) => {
    const result = await agent.invoke({ messages: [{ role: "user", content: query }] });
    ls.logOutputs(result);

    // 检查流程是否为 HUMAN -> AI FINAL RESPONSE（未调用工具）
    expect(result.messages).toHaveLength(numMessages);
  }
);

测试 2：简单工具调用

对于工具调用，我们将验证代理是否使用正确的参数调用了正确的工具。

@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

ls.test(
  "should search for correct ticker",
  {
    inputs: { query: "What is the price of Apple?" },
    expected: { numMessages: 4 },
  },
  async ({ inputs: { query }, expected: { numMessages } }) => {
    const result = await agent.invoke({
      messages: [{ role: "user", content: query }],
    });

    ls.logOutputs(result);

    // 代理应该对股票代码工具进行了一次工具调用
    const toolCalls = (result.messages[1] as AIMessage).tool_calls || [];
    const tickerQuery = JSON.parse(toolCalls[0].function.arguments).query.ticker;
    // 检查是否查询了正确的股票代码
    expect(tickerQuery).toBe("AAPL");

    // 检查流程是否为 HUMAN -> AI -> TOOL -> AI FINAL RESPONSE
    expect(result.messages).toHaveLength(numMessages);
  }
);

ls.test(
  "should search for correct ticker",
  {
    inputs: { query: "What is the price of Apple?" },
    expected: { numMessages: 4 },
  },
  async ({ inputs: { query }, expected: { numMessages } }) => {
    const result = await agent.invoke({
      messages: [{ role: "user", content: query }],
    });

    ls.logOutputs(result);

    // 代理应该对股票代码工具进行了一次工具调用
    const toolCalls = (result.messages[1] as AIMessage).tool_calls || [];
    const tickerQuery = JSON.parse(toolCalls[0].function.arguments).query.ticker;
    // 检查是否查询了正确的股票代码
    expect(tickerQuery).toBe("AAPL");

    // 检查流程是否为 HUMAN -> AI -> TOOL -> AI FINAL RESPONSE
    expect(result.messages).toHaveLength(numMessages);
  }
);

测试 3：复杂工具调用

有些工具调用比其他调用更容易测试。对于股票代码查找，我们可以断言搜索了正确的股票代码。对于编码工具，工具的输入和输出约束较少，并且有很多方法可以得到正确答案。在这种情况下，更简单的方法是通过运行完整的代理并断言它既调用了编码工具又最终得到了正确答案来测试工具是否被正确使用。

@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

ls.test(
  "should execute code when needed",
  {
    inputs: { query: "What was the average return rate for FAANG stock in 2024?" },
    expected: { answer: 53 },
  },
  async ({ inputs: { query }, expected: { answer } }) => {
    const result = await agent.invoke({
      messages: [{ role: "user", content: query }],
    });

    ls.logOutputs(result);

    // 获取 LLM 进行的所有工具调用
    const toolCalls = result.messages
      .filter(m => (m as AIMessage).tool_calls)
      .flatMap(m => (m as AIMessage).tool_calls?.map(tc => tc.name));
    // 这将记录 LLM 采取的步骤数，我们可以随时间跟踪以衡量性能
    ls.logFeedback({
      key: "num_steps",
      score: result.messages.length - 1, // 第一条消息是用户消息
    });
    // 断言工具调用包含 "code_tool" 函数
    expect(toolCalls).toContain("code_tool");
    // 断言答案与预期答案相差在 1 以内
    expect(Math.abs(result.structured_response.numeric_answer - answer)).toBeLessThanOrEqual(1);
  }
);

ls.test(
  "should execute code when needed",
  {
    inputs: { query: "What was the average return rate for FAANG stock in 2024?" },
    expected: { answer: 53 },
  },
  async ({ inputs: { query }, expected: { answer } }) => {
    const result = await agent.invoke({
      messages: [{ role: "user", content: query }],
    });

    ls.logOutputs(result);
    // 获取 LLM 进行的所有工具调用
    const toolCalls = result.messages
      .filter(m => (m as AIMessage).tool_calls)
      .flatMap(m => (m as AIMessage).tool_calls?.map(tc => tc.name));
    // 这将记录 LLM 采取的步骤数，我们可以随时间跟踪以衡量性能
    ls.logFeedback({
      key: "num_steps",
      score: result.messages.length - 1, // 第一条消息是用户消息
    });
    // 断言工具调用包含 "code_tool" 函数
    expect(toolCalls).toContain("code_tool");
    // 断言答案与预期答案相差在 1 以内
    expect(Math.abs(result.structured_response.numeric_answer - answer)).toBeLessThanOrEqual(1);
  }
);

测试 4：LLM 作为评判者

我们将通过运行 LLM 作为评判者评估来确保代理的答案基于搜索结果。为了将 LLM 作为评判者的调用与我们的代理分开跟踪，我们将在 Python 中使用 LangSmith 提供的 trace_feedback 上下文管理器，在 JS/TS 中使用 wrapEvaluator 函数。

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']

// 此代码放在测试外部 - 它只是一个辅助函数
const judgeLLM = new ChatOpenAI({ model: "gpt-5.4" });
const groundedEvaluator = async (params: {
  answer: string;
  referenceDocuments: string,
}) => {
  // LLM 评判者的指令
  const instructions = [
    "Return 1 if the ANSWER is grounded in the DOCUMENTS",
    "Return 0 if the ANSWER is not grounded in the DOCUMENTS",
  ].join("\n");

  // 运行评判 LLM
  const grade = await judgeLLM.invoke([
    { role: "system", content: instructions },
    { role: "user", content: `ANSWER: ${params.answer}\nDOCUMENTS: ${params.referenceDocuments}` },
  ]);
  const score = parseInt(grade.content.toString());
  return { key: "groundedness", score };
};

// 此代码放在测试内部
ls.test(
  "grounded in the source",
  {
    inputs: { query: "How did Nvidia stock do in 2024 according to analysts?" },
  },
  async ({ inputs: { query } }) => {
    const result = await agent.invoke({
      messages: [{ role: "user", content: query }],
    });
    const wrappedEvaluator = ls.wrapEvaluator(groundedEvaluator);
    await wrappedEvaluator({
      answer: result.structuredResponse.text_answer ?? "",
      referenceDocuments: result.structuredResponse.reasoning,
    })
    ls.logOutputs(result);
  }
);

// 此代码放在测试外部 - 它只是一个辅助函数
const judgeLLM = new ChatOpenAI({ model: "gpt-5.4" });
const groundedEvaluator = async (params: {
  answer: string;
  referenceDocuments: string,
}) => {
  // LLM 评判者的指令
  const instructions = [
    "Return 1 if the ANSWER is grounded in the DOCUMENTS",
    "Return 0 if the ANSWER is not grounded in the DOCUMENTS",
  ].join("\n");

  // 运行评判 LLM
  const grade = await judgeLLM.invoke([
    { role: "system", content: instructions },
    { role: "user", content: `ANSWER: ${params.answer}\nDOCUMENTS: ${params.referenceDocuments}` },
  ]);
  const score = parseInt(grade.content.toString());
  return { key: "groundedness", score };
};

// 此代码放在测试内部
ls.test(
  "grounded in the source",
  {
    inputs: { query: "How did Nvidia stock do in 2024 according to analysts?" },
  },
  async ({ inputs: { query } }) => {
    const result = await agent.invoke({
      messages: [{ role: "user", content: query }],
    });
    const wrappedEvaluator = ls.wrapEvaluator(groundedEvaluator);
    await wrappedEvaluator({
      answer: result.structuredResponse.text_answer ?? "",
      referenceDocuments: result.structuredResponse.reasoning,
    })
    ls.logOutputs(result);
  }
);

运行测试

一旦您设置了配置文件（如果您使用 Vitest 或 Jest），您可以使用以下命令运行测试：

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"],
    testTimeout: 30000,
  },
});

创建一个 `ls.jest.config.ts` 文件：

require('dotenv').config();

module.exports = {
  preset: 'ts-jest',
  testEnvironment: 'node',
  testMatch: [
    '<rootDir>/tests/jest/**/*.jest.eval.ts'
  ],
  testPathIgnorePatterns: [
    '<rootDir>/tests/vitest/.*.vitest.eval.ts$'
  ],
  reporters: ["langsmith/jest/reporter"],
  testTimeout: 30000,
};

pytest --langsmith-output tests

yarn vitest --config ls.vitest.config.ts

yarn jest --config ls.jest.config.ts

参考代码

请记住将 Vitest 和 Jest 的配置文件也添加到您的项目中。

代理

代理代码

from e2b_code_interpreter import Sandbox
from langchain_community.tools import PolygonAggregates, TavilySearchResults
from langchain_community.utilities.polygon import PolygonAPIWrapper
from langchain.agents import create_agent
from typing_extensions import Annotated, TypedDict


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}"

polygon_aggregates = PolygonAggregates(api_wrapper=PolygonAPIWrapper())

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="您是一位金融专家。请准确回答用户的问题",
)

import { ChatOpenAI } from "@langchain/openai";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { Sandbox } from '@e2b/code-interpreter'
import { restClient } from '@polygon.io/client-js';
import { tool } from "@langchain/core/tools";
import { z } from "zod";

const codeTool = tool(async (input) => {
    const sbx = await Sandbox.create();
    const execution = await sbx.runCode(input.code);
    if (execution.error) {
    return `Error: ${execution.error}`;
    }
    return `Results: ${execution.results}, Logs: ${execution.logs}`;
}, {
    name: "code",
    description: "执行 Python 代码并返回结果。",
    schema: z.object({
    code: z.string().describe("要执行的 Python 代码"),
    }),
});

const TickerToolInputSchema = z.object({
    ticker: z.string().describe("股票的代码符号"),
    timespan: z.enum(["minute", "hour", "day", "week", "month", "quarter", "year"]).describe("时间窗口的大小。"),
    timespan_multiplier: z.number().describe("时间窗口的乘数"),
    from_date: z
    .string()
    .describe("开始拉取数据的日期，YYYY-MM-DD 格式 - 仅包含年月日"),
    to_date: z
    .string()
    .describe("停止拉取数据的日期，YYYY-MM-DD 格式 - 仅包含年月日"),
});

const rest = restClient(process.env.POLYGON_API_KEY);

const tickerTool = tool(async (query) => {
    const parsed = TickerToolInputSchema.parse(query);
    const result = await rest.stocks.aggregates(
    parsed.ticker,
    parsed.timespan_multiplier,
    parsed.timespan,
    parsed.from_date,
    parsed.to_date
    );
    return JSON.stringify(result);
}, {
    name: "ticker",
    description: "拉取股票代码的数据",
    schema: TickerToolInputSchema,
});

const searchTool = new TavilySearchResults({
    maxResults: 5,
});

const AgentOutputFormatSchema = z.object({
    numeric_answer: z.number().optional().describe("数值答案，如果用户要求提供"),
    text_answer: z.string().optional().describe("文本答案，如果用户要求提供"),
    reasoning: z.string().describe("答案背后的推理过程"),
})

const tools = [codeTool, searchTool, tickerTool];

const agent = createReactAgent({
    llm: new ChatOpenAI({ model: "gpt-5.4" }),
    tools: tools,
    responseFormat: AgentOutputFormatSchema,
    stateModifier: "您是一位金融专家。请准确回答用户的问题",
});

export default agent;

测试

测试代码

# from app import agent, polygon_aggregates, search_tool # 从您的代理定义位置导入
import pytest
from langchain.chat_models import init_chat_model
from langsmith import testing as t
from typing_extensions import Annotated, TypedDict

@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

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"]

import { expect } from "vitest";
import * as ls from "langsmith/vitest";
import agent from "../agent";
import { AIMessage, ToolMessage } from "@langchain/core/messages";
import { ChatOpenAI } from "@langchain/openai";

const judgeLLM = new ChatOpenAI({ model: "gpt-5.4" });

const groundedEvaluator = async (params: {
  answer: string;
  referenceDocuments: string,
}) => {
  const instructions = [
    "Return 1 if the ANSWER is grounded in the DOCUMENTS",
    "Return 0 if the ANSWER is not grounded in the DOCUMENTS",
  ].join("\n");

  const grade = await judgeLLM.invoke([
    { role: "system", content: instructions },
    { role: "user", content: `ANSWER: ${params.answer}\nDOCUMENTS: ${params.referenceDocuments}` },
  ]);
  const score = parseInt(grade.content.toString());
  return { key: "groundedness", score };
};

ls.describe("Agent Tests", () => {
  ls.test.each([
    { inputs: { query: "Hello!" }, referenceOutputs: { numMessages: 2 } },
    { inputs: { query: "How are you doing?" }, referenceOutputs: { numMessages: 2 } },
  ])(
    "should not use tools on offtopic query: %s",
    async ({ inputs: { query }, referenceOutputs: { numMessages } }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      ls.logOutputs(result);
      expect(result.messages).toHaveLength(numMessages);
    }
  );

  ls.test(
    "should search for correct ticker",
    {
      inputs: { query: "What is the price of Apple?" },
      referenceOutputs: { numMessages: 4 },
    },
    async ({ inputs: { query }, referenceOutputs: { numMessages } }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      const toolCalls = (result.messages[1] as AIMessage).tool_calls || [];
      const tickerQuery = toolCalls[0].args.ticker;
      ls.logOutputs(result);
      expect(tickerQuery).toBe("AAPL");
      expect(result.messages).toHaveLength(numMessages);
    }
  );

  ls.test(
    "should execute code when needed",
    {
      inputs: { query: "What was the average return rate for FAANG stock in 2024?" },
      referenceOutputs: { answer: 53 },
    },
    async ({ inputs: { query }, referenceOutputs: { answer } }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });

      const toolCalls = result.messages
        .filter(m => (m as AIMessage).tool_calls)
        .flatMap(m => (m as AIMessage).tool_calls?.map(tc => tc.name));
      ls.logFeedback({
        key: "num_steps",
        score: result.messages.length - 1,
      });
      ls.logOutputs(result);
      expect(toolCalls).toContain("code_tool");
      expect(Math.abs((result.structuredResponse.numeric_answer ?? 0) - answer)).toBeLessThanOrEqual(1);
    }
  );

  ls.test(
    "grounded in the source",
    {
      inputs: { query: "How did Nvidia stock do in 2024?" },
      referenceOutputs: {},
    },
    async ({ inputs: { query }, referenceOutputs: {} }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      const referenceDocuments = result.messages
        .filter((m): m is ToolMessage => m.name?.includes('tavily_search_results_json') ?? false)
        .map(m => m.content)
        .join('\n');
      const wrappedEvaluator = ls.wrapEvaluator(groundedEvaluator);
      await wrappedEvaluator({
        answer: result.structuredResponse.text_answer ?? "",
        referenceDocuments: referenceDocuments,
      })
      ls.logOutputs(result);
    }
  );
});

import { expect } from "@jest/globals";
import * as ls from "langsmith/jest";
import agent from "../agent";
import { AIMessage } from "@langchain/core/messages";
import { ChatOpenAI } from "@langchain/openai";

const judgeLLM = new ChatOpenAI({ model: "gpt-5.4" });

const groundedEvaluator = async (params: {
  answer: string;
  referenceDocuments: string,
}) => {
  const instructions = [
    "Return 1 if the ANSWER is grounded in the DOCUMENTS",
    "Return 0 if the ANSWER is not grounded in the DOCUMENTS",
  ].join("\n");

  const grade = await judgeLLM.invoke([
    { role: "system", content: instructions },
    { role: "user", content: `ANSWER: ${params.answer}\nDOCUMENTS: ${params.referenceDocuments}` },
  ]);
  const score = parseInt(grade.content.toString());
  return { key: "groundedness", score };
};

ls.describe("Agent Tests", () => {
  ls.test.each([
    { inputs: { query: "Hello!" }, referenceOutputs: { numMessages: 2 } },
    { inputs: { query: "How are you doing?" }, referenceOutputs: { numMessages: 2 } },
  ])(
    "should not use tools on offtopic query: %s",
    async ({ inputs: { query }, referenceOutputs: { numMessages } }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      ls.logOutputs(result);
      expect(result.messages).toHaveLength(numMessages);
    }
  );

  ls.test(
    "should search for correct ticker",
    {
      inputs: { query: "What is the price of Apple?" },
      referenceOutputs: { numMessages: 4 },
    },
    async ({ inputs: { query }, referenceOutputs: { numMessages } }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      const toolCalls = (result.messages[1] as AIMessage).tool_calls || [];
      const tickerQuery = toolCalls[0].args.ticker;
      ls.logOutputs(result);
      expect(tickerQuery).toBe("AAPL");
      expect(result.messages).toHaveLength(numMessages);
    }
  );

  ls.test(
    "should execute code when needed",
    {
      inputs: { query: "What was the average return rate for FAANG stock in 2024?" },
      referenceOutputs: { answer: 53 },
    },
    async ({ inputs: { query }, referenceOutputs: { answer } }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      const toolCalls = result.messages
        .filter(m => (m as AIMessage).tool_calls)
        .flatMap(m => (m as AIMessage).tool_calls?.map(tc => tc.name));
      ls.logFeedback({
        key: "num_steps",
        score: result.messages.length - 1,
      });
      ls.logOutputs(result);
      expect(toolCalls).toContain("code_tool");
      expect(Math.abs((result.structuredResponse.numeric_answer ?? 0) - answer)).toBeLessThanOrEqual(1);
    }
  );

  ls.test(
    "grounded in the source",
    {
      inputs: { query: "How did Nvidia stock do in 2024 according to analysts?" },
      referenceOutputs: {},
    },
    async ({ inputs: { query }, referenceOutputs: {} }) => {
      const result = await agent.invoke({
        messages: [{ role: "user", content: query }],
      });
      const wrappedEvaluator = ls.wrapEvaluator(groundedEvaluator);
      await wrappedEvaluator({
        answer: result.structuredResponse.text_answer ?? "",
        referenceDocuments: result.structuredResponse.reasoning,
      })
      ls.logOutputs(result);
    }
  );
});

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面或提交问题。

Datasets

Set up evaluations

Analyze experiment results

Annotation & human feedback

Common data types

使用 Pytest/Vitest 和 LangSmith 测试 ReAct 代理

设置

安装

环境变量

创建您的应用

定义工具

定义代理

编写测试

测试 1：处理无关问题

测试 2：简单工具调用

测试 3：复杂工具调用

测试 4：LLM 作为评判者

运行测试

参考代码

代理

测试

​设置

​安装

​环境变量

​创建您的应用

​定义工具

​定义代理

​编写测试

​测试 1：处理无关问题

​测试 2：简单工具调用

​测试 3：复杂工具调用

​测试 4：LLM 作为评判者

​运行测试

​参考代码

​代理

​测试

设置

安装

环境变量

创建您的应用

定义工具

定义代理

编写测试

测试 1：处理无关问题

测试 2：简单工具调用

测试 3：复杂工具调用

测试 4：LLM 作为评判者

运行测试

参考代码

代理

测试