Skip to content

如何在本地测试LangGraph应用

本指南假设您已经正确设置了LangGraph应用，并且有一个合适的配置文件和相应的编译后的图，同时您还拥有了一个有效的LangChain API密钥。

在本地进行测试可以确保没有错误或Python依赖项冲突，并确认配置文件指定正确。

设置

安装LangGraph CLI包：

pip install -U "langgraph-cli[inmem]"

确保你有一个API密钥，你可以在LangSmith UI（设置 > API密钥）中创建。这是为了验证你有LangGraph Cloud访问权限。在你将密钥保存到安全位置后，在你的.env文件中添加以下行：

LANGSMITH_API_KEY = *********

启动API服务器

安装CLI后，您可以运行以下命令来启动API服务器，以便进行本地测试：

langgraph dev

这将启动LangGraph API服务器进行本地测试。如果成功运行，您应该会看到类似以下内容的信息：

准备就绪！

内存模式

langgraph dev 命令以内存模式启动LangGraph服务器。此模式适用于开发和测试目的。对于生产环境，您应该部署LangGraph服务器并提供持久存储后端的访问权限。

如果您希望使用持久存储后端测试应用程序，可以使用 langgraph up 命令而不是 langgraph dev。您需要在机器上安装 docker 才能使用此命令。

与服务器交互

现在，我们可以使用LangGraph SDK与API服务器进行交互。首先，我们需要启动客户端，并选择我们的助手（在这种情况下，我们选择了一个名为“agent”的图，确保选择您要测试的正确助手）。

您可以使用传递身份验证或设置环境变量的方式初始化。

使用身份验证初始化

from langgraph_sdk import get_client

# 只有在调用 langgraph dev 时更改默认端口的情况下才将 url 传递给 get_client()
client = get_client(url=<DEPLOYMENT_URL>,api_key=<LANGSMITH_API_KEY>)
# 使用名为 "agent" 的图
assistant_id = "agent"
thread = await client.threads.create()

import { Client } from "@langchain/langgraph-sdk";

// 只有在调用 langgraph dev 时更改默认端口的情况下才设置 apiUrl
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <LANGSMITH_API_KEY> });
// 使用名为 "agent" 的图
const assistantId = "agent";
const thread = await client.threads.create();

curl --request POST \
  --url <DEPLOYMENT_URL>/threads \
  --header 'Content-Type: application/json'
  --header 'x-api-key: <LANGSMITH_API_KEY>'

使用环境变量初始化

如果您已经在环境中设置了 LANGSMITH_API_KEY，则不需要显式传递身份验证给客户端

from langgraph_sdk import get_client

# 只有在调用 langgraph dev 时更改默认端口的情况下才将 url 传递给 get_client()
client = get_client()
# 使用名为 "agent" 的图
assistant_id = "agent"
thread = await client.threads.create()

import { Client } from "@langchain/langgraph-sdk";

// 只有在调用 langgraph dev 时更改默认端口的情况下才设置 apiUrl
const client = new Client();
// 使用名为 "agent" 的图
const assistantId = "agent";
const thread = await client.threads.create();

curl --request POST \
  --url <DEPLOYMENT_URL>/threads \
  --header 'Content-Type: application/json'

现在，我们可以调用我们的图以确保其正常工作。请确保更改输入以匹配图的正确模式。

input = {"messages": [{"role": "user", "content": "what's the weather in sf"}]}
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id,
    input=input,
    stream_mode="updates",
):
    print(f"接收到新事件类型: {chunk.event}...")
    print(chunk.data)
    print("\n\n")

const input = { "messages": [{ "role": "user", "content": "what's the weather in sf"}] }

const streamResponse = client.runs.stream(
  thread["thread_id"],
  assistantId,
  {
    input: input,
    streamMode: "updates",
  }
);
for await (const chunk of streamResponse) {
  console.log(`接收到新事件类型: ${chunk.event}...`);
  console.log(chunk.data);
  console.log("\n\n");
}

curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"input\": {\"messages\": [{\"role\": \"human\", \"content\": \"what's the weather in sf\"}]},
   \"stream_mode\": [
     \"events\"
   ]
 }" | \
 sed 's/\r$//' | \
 awk '
 /^event:/ {
     if (data_content != "") {
         print data_content "\n"
     }
     sub(/^event: /, "Receiving event of type: ", $0)
     printf "%s...\n", $0
     data_content = ""
 }
 /^data:/ {
     sub(/^data: /, "", $0)
     data_content = $0
 }
 END {
     if (data_content != "") {
         print data_content "\n"
     }
 }
 '

如果您的图正常工作，您应该会在控制台中看到图的输出。当然，还有许多其他方式可以测试您的图，有关可以发送的完整命令列表，请参阅Python和JS/TS参考文档。

Comments