运行 托管深度代理 涉及创建线程、启动运行和流式传输输出。 部署代理 首先,然后使用 SDK 或 REST API 调用它。
如需更短的端到端路径,请参阅 快速入门.
设置
为您的运行时安装 SDK:
uv add managed-deepagents
# Or with pip:
pip install managed-deepagents
npm install @langchain/managed-deepagents
设置请求默认值:
SDK 默认读取 LANGSMITH_API_KEY 。REST 请求需要 X-Api-Key header:
X-Api-Key:
查找代理 ID
使用 CLI、SDK 或 REST API 查找代理 ID。
CLI
需要 deepagents-cli。请参阅 安装客户端.
列出代理:
deepagents agents list
检查一个代理:
deepagents agents get <agent_id>
SDK
列出代理并读取 id 您想要的代理的:
from managed_deepagents import Client
with Client() as client:
agents = client.agents.list()
print("Existing agents:")
for item in agents["items"]:
print(f" {item['id']} {item['name']}")
const client = new Client();
const agents = await client.agents.list();
console.log("Existing agents:");
for (const item of agents.items ?? []) {
console.log(` ${item.id} ${item.name}`);
}
检查一个代理:
from managed_deepagents import Client
agent_id = "<agent_id>"
with Client() as client:
# include_files=True returns file and skill contents, not just metadata.
agent = client.agents.get(agent_id, include_files=True)
print(json.dumps(agent, indent=2))
const client = new Client();
const agentId = "<agent_id>";
// includeFiles: true returns file and skill contents, not just metadata.
const agent = await client.agents.get(agentId, { includeFiles: true });
console.log(JSON.stringify(agent, null, 2));
API
使用 GET /v1/deepagents/agents列出代理,然后读取 id 您想要的代理的:
curl --request GET \
--url "$DEEPAGENTS_BASE_URL/agents" \
--header "X-Api-Key: $LANGSMITH_API_KEY"
要检查单个代理,请调用 GET /v1/deepagents/agents/{agent_id}.
创建线程并流式传输运行
您可以使用 SDK 或 REST API 创建线程并流式传输运行。目前没有用于运行托管深度代理的 CLI 命令。
中检索到的代理 ID。以下示例重用这些变量:
Python SDK
from managed_deepagents import Client
agent_id = os.environ["AGENT_ID"]
client = Client()
TypeScript SDK
const agentId = process.env.AGENT_ID!;
const client = new Client();
cURL
# The cURL examples below reuse the shell variables exported above.
# No additional setup is required.
创建线程
在运行代理之前创建线程。线程为长时间运行的工作保留对话和执行状态。
options 对象是可选的,两个字段默认为 false。设置 test_run to true 将线程标记为测试运行,该运行会被过滤掉使用量和分析数据。默认情况下, skip_memory_write_protection 让运行时在代理写入长期记忆之前引发人工介入中断,以便您可以批准或拒绝写入。将其设置为 true 让记忆写入立即进行,这在没有人员可用以批准写入的无头运行中很有用。有关完整字段参考,请参阅 API 参考.
Python SDK
thread = client.threads.create(
agent_id=agent_id,
options={
"test_run": False,
"skip_memory_write_protection": False,
},
)
thread_id = thread["id"]
print(f"Thread ID: {thread_id}")
TypeScript SDK
const thread = await client.threads.create({
agent_id: agentId,
options: {
test_run: false,
skip_memory_write_protection: false,
},
});
const threadId = thread.id;
console.log(`Thread ID: ${threadId}`);
cURL
curl --request POST \
--url "$DEEPAGENTS_BASE_URL/threads" \
--header "X-Api-Key: $LANGSMITH_API_KEY" \
--header 'Content-Type: application/json' \
--data '{
"agent_id": "'"$AGENT_ID"'",
"options": {
"test_run": false,
"skip_memory_write_protection": false
}
}'
在流式传输之前设置返回的线程 ID:
从线程流式传输运行
在线程上开始工作并流式传输结果:
Python SDK
for event in client.threads.stream(
thread_id,
agent_id=agent_id,
messages=[
{
"role": "user",
"content": "Research recent approaches to agent memory and summarize the main trade-offs.",
}
],
stream_mode=["values", "updates", "messages-tuple"],
stream_subgraphs=True,
user_timezone="America/Los_Angeles",
):
print(event.event, event.data)
TypeScript SDK
const langGraphClient = client.getLangGraphClient({ agentId });
const stream = langGraphClient.runs.stream(threadId, agentId, {
input: {
messages: [
{
role: "user",
content:
"Research recent approaches to agent memory and summarize the main trade-offs.",
},
],
},
streamMode: ["values", "updates", "messages-tuple"],
streamSubgraphs: true,
});
for await (const event of stream) {
console.log(event.event, event.data);
}
cURL
curl --request POST \
--url "$DEEPAGENTS_BASE_URL/threads/$THREAD_ID/runs/stream" \
--header "X-Api-Key: $LANGSMITH_API_KEY" \
--header 'Accept: text/event-stream' \
--header 'Content-Type: application/json' \
--data '{
"agent_id": "'"$AGENT_ID"'",
"input": {
"messages": [
{
"role": "user",
"content": "Research recent approaches to agent memory and summarize the main trade-offs."
}
]
},
"stream_mode": ["values", "updates", "messages-tuple"],
"stream_subgraphs": true,
"user_timezone": "America/Los_Angeles"
}'
该端点流式传输 Server-Sent Events (text/event-stream)。使用流式传输模式 (stream_mode, streamMode) 时,您会收到增量 updates 和 messages-tuple 事件,随着代理工作,最后一个 values 事件包含运行的完整状态,包括代理的响应。设置 stream_subgraphs (streamSubgraphs) to true 也可以流式传输子图(如子代理)的事件。可选的 user_timezone 设置调用者的 IANA 时区,以便代理以本地时间推理日期,默认为代理配置的时区或 UTC.
cURL 示例打印原始 SSE 行。解析其 data: 负载为 JSON 以驱动 UI。Python 和 TypeScript SDK 示例使用 event.event 和 event.data.
选择流式传输模式:
| 流式传输模式 | 用途 |
|---|---|
values | 步骤后的完整状态快照。 |
updates | 代理工作时的增量状态更新。 |
messages-tuple | 用于聊天 UI 的令牌级消息输出。发出一个 messages 事件,其负载是一个 [chunk, metadata] 元组。 |
使用 React 进行流式传输 useStream
前面的 Python SDK 和 TypeScript SDK 示例流式传输路由级事件。以下 React useStream 示例公开了 LangGraph 投影,例如 stream.messages, stream.values,以及用于聊天 UI 的输出状态。
对于 React 应用程序,请将 TypeScript SDK 的 LangGraph 客户端适配器与 @langchain/react:
const agentId = "<agent_id>";
const managedDeepAgents = new Client();
const client = managedDeepAgents.getLangGraphClient({ agentId });
const stream = useStream({
client,
assistantId: agentId
});
return (
<section>
<button
type="button"
disabled={stream.isLoading}
onClick={() => {
void stream.submit({
messages: [
{ role: "user", content: "Write a short status update." },
],
});
}}
>
Run agent
</button>
{stream.messages.map((message, index) => (
<p key={message.id ?? index}>{String(message.content)}</p>
))}
</section>
);
}
后续步骤
SDK reference
Python 和 TypeScript 客户端的 SDK 配置详情。
API reference
路由级请求和响应详情。