跳到内容

如何在本地测试 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 Server。此模式适用于开发和测试目的。对于生产用途,您应该部署 LangGraph Server 并访问持久存储后端。

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

与服务器交互

现在我们可以使用 LangGraph SDK 与 API 服务器交互。首先,我们需要启动客户端,选择我们的助手(在此例中是一个名为“agent”的图,请确保选择您希望测试的合适助手)。

您可以通过传递认证信息或设置环境变量进行初始化。

使用认证进行初始化

from langgraph_sdk import get_client

# only pass the url argument to get_client() if you changed the default port when calling langgraph dev
client = get_client(url=<DEPLOYMENT_URL>,api_key=<LANGSMITH_API_KEY>)
# Using the graph deployed with the name "agent"
assistant_id = "agent"
thread = await client.threads.create()
import { Client } from "@langchain/langgraph-sdk";

// only set the apiUrl if you changed the default port when calling langgraph dev
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <LANGSMITH_API_KEY> });
// Using the graph deployed with the name "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

# only pass the url argument to get_client() if you changed the default port when calling langgraph dev
client = get_client()
# Using the graph deployed with the name "agent"
assistant_id = "agent"
thread = await client.threads.create()
import { Client } from "@langchain/langgraph-sdk";

// only set the apiUrl if you changed the default port when calling langgraph dev
const client = new Client();
// Using the graph deployed with the name "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"Receiving new event of type: {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(`Receiving new event of type: ${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"
     }
 }
 ' 

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

评论