Skip to content

如何在本地测试LangGraph应用

本指南假设您已经正确设置了LangGraph应用,并且有一个适当的配置文件以及相应的编译后的图谱,并且您拥有一个有效的LangChain API密钥。

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

设置

安装 LangGraph CLI 包:

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

确保您有一个 API 密钥,您可以在 LangSmith UI(设置 > API 密钥)中创建它。此密钥用于验证您具有 LangGraph 云访问权限。在将密钥保存到安全位置后,在您的 .env 文件中添加以下行:

LANGSMITH_API_KEY = *********

启动API服务器

安装CLI后,您可以运行以下命令来启动用于本地测试的API服务器:

langgraph dev

这将在本地启动LangGraph API服务器。如果成功运行,您应该会看到类似以下的内容:

准备好了!

内存模式

langgraph dev 命令以内存模式启动LangGraph Server。此模式适用于开发和测试目的。在生产环境中使用时,您应该部署LangGraph Server并使其能够访问持久存储后端。

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

与服务器交互

现在我们可以使用LangGraph SDK与API服务器进行交互。首先,我们需要启动客户端,并选择我们的助手(在这种情况下,我们称之为“agent”的图,请确保选择您要测试的正确助手)。

您可以通过传递身份验证或设置环境变量来进行初始化。

使用身份验证初始化

from langgraph_sdk import get_client

# 只有在调用langgraph dev时更改默认端口的情况下才向get_client()传递url参数
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时更改默认端口的情况下才向get_client()传递url参数
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"
     }
 }
 ' 

如果您的图正确工作,您应该会在控制台中看到图的输出。当然,还有许多其他方式可能需要测试您的图,有关您可以发送的完整命令列表,请参阅PythonJS/TS 参考资料。

Comments