对于 AI 代理:可在 https://www.mongodb.com/zh-cn/docs/llms.txt 获取文档索引—通过在任何 URL 路径后添加 .md 可获取所有页面的 Markdown 版本。
See how MongoDB 9.0 delivers up to 2x higher throughput.
MongoDB Branding Shape
Register now >
Docs 菜单

Agent-engine-sdk-langgraph

适用于MongoDB Atlas Agent Engine 的 LangChain SDK。通过 agent-engine-runner-shared 平台运行时提供特定于 LangChain 的瘦包装器。

pip install agent-engine-sdk-langgraph

或者在 uv项目中:

uv add agent-engine-sdk-langgraph
from agent_engine_sdk_langgraph import App
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
app = App(app_name="my-agent", app_version="1.0.0")
@app.tool()
def lookup(query: str) -> str:
"""Search the knowledge base."""
return "result for " + query
@app.entrypoint
def build_agent():
from langchain_openai import ChatOpenAI
llm = app.llm(ChatOpenAI(model="gpt-5.4"))
tools = app.get_tools()
def call_model(state: MessagesState):
response = llm.invoke(state["messages"])
return {"messages": [response]}
graph = StateGraph(MessagesState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
graph.set_entry_point("agent")
graph.add_edge("tools", "agent")
return graph.compile(checkpointer=app.checkpointer())
app.run()

app.memory 是统一的 `agent-engine-sdk-memory <../agent-engine-sdk-memory/README.md>`__ Memory 门面(在平台运行时上受应用程序限制)。每次来自参数、绑定上下文或环境执行上下文的调用时,都会解析身份。

app = App(app_name="my-agent")
# Save a semantic fact — returns CreateSemanticResult
result = app.memory.save_semantic(
text="User prefers dark mode",
label="pref-theme",
user_id="u1",
metadata={"channel": "web", "priority": "high"}, # optional caller-supplied metadata
)
if result.acknowledged:
...
# Search — returns list[MemoryChunk]
chunks = app.memory.search_semantic(query="user preferences", user_id="u1")
for chunk in chunks:
print(chunk.content)
# Build prompt context — returns ContextResponse
# Default sources are LTM (episodic + semantic); pass
# enabled_sources={"stm", ...} to include recent turns.
context = app.memory.build_context(query="help me", user_id="u1")
prompt_block = context.formatted_context # str (or structured list, depending on format)

build_context_from_sources (按源检索模式、筛选器和 top_k)在环境 app.memory 运行时上可用。它返回完整的 ContextResponse,因此每个源的元数据(ranking_strategy、source_outcomes) 会保留下来;平台会对租户进行标记,并通过持久性执行将响应传回。仅当请求 stm 源时才需要 session_id。

from agent_engine_sdk_langgraph import Memory 重新导出与 agent_engine_sdk_memory.Memory 相同的类。自行构建 Memory(api_key=...) / Memory(base_url=...) 是HTTP/direct 路径 — 不受环境应用限制。

当这个问题发生时

这是一个难题:app.memory 上没有双API ,也没有兼容性填充程序。客户数代理仅在重建代理映像时中断,该映像拾取包含此 SDK 的基于 Runner 的轮子。仅进行平台合并或重新部署旧代理映像不会更改已嵌入该映像中的 SDK 代码。不存在存储内存数据迁移— 只有客户端返回形状和调用约定发生变化。

从 pre-facade 表面迁移

  • 返回类型/真实性/元数据/``build_context``:使用 acknowledged 写入返回类型化结果(CreateSemanticResult、CreateEpisodicResult 等),而不是裸露的 bool/字典。首选 if result.acknowledged:(而非 if result: — Pydantic 模型始终为 true)。相似性搜索返回 list[MemoryChunk](chunk.content,可选 chunk.similarity_score)。曾经是顶级字典键的松散字段(title,summary,tags,term,definition,related_terms,...)位于 chunk.metadata 下(例如 ep.metadata.get("title") 而不是ep.get("title"))。 build_context 返回 ContextResponse;使用 context.formatted_context,而不是字符串形式的返回值。

  • 写入助手只能使用关键字 (save_semantic(text=..., label=..., …))。位置调用引发 TypeError。

  • 使用“visibility="private"”进行读取会保留环境用户过滤(之前传递任何 visibility 参数都会删除该筛选器):私有可见性搜索现在仅返回当前用户的记忆,除非传递了显式 user_id。

  • ``top_k``:每个源的 search_semantic/search_episodes/search_taxonomic默认为 top_k=50(通常为 10)。 discover_procedures 仍默认为 10。统一 Memory.search() 仍默认为 top_k=10。公共 build_context 没有 top_k 参数。使用 max_tokens 作为上下文构建总预算(而不是获取费用)。在检索和排名后,服务器减去 500-令牌格式保留,然后贪心地选择适合余数的整个内存块。等于或低于 500 的正值不会留下内存预算。当没有适合的数据块时,高于 500 的值仍可能产生空上下文。如果需要旧限制,请在搜索助手上传递显式 top_k。

  • 身份:无法解析的必填字段引发 MemoryIdentityError(不再有软 None/静默跳过)。 save_episode 需要可解析的 session_id(环境调用上下文即可;否则会显式传递或绑定 MemoryRequestContext)。空白值不算作设立。

  • 应用程序绑定的创建保真度:创建结果 id 可能是 "",has_embedding 通常是 False — 在 .acknowledged(而不是 id)上启动成功。每个操作的详细信息位于内存包功能矩阵中。

  • 获取与搜索: get_semantic /get_taxonomic_term / list_episodes 在应用绑定时仍会返回松散类型的字典。只有 search* 方法会返回 list[MemoryChunk]。

  • 重命名:如果有人调用了外观前名称,请使用外观公共API — create_taxonomic → save_taxonomic; list_taxonomic_domains → list_domains。

之前/之后

# save_semantic: bool → .acknowledged; positional → keyword-only
# before
ok = app.memory.save_semantic("User prefers dark mode", "pref-theme", user_id="u1")
if ok:
...
# after
result = app.memory.save_semantic(
text="User prefers dark mode",
label="pref-theme",
user_id="u1",
)
if result.acknowledged:
...
# save_episode: str|None → CreateEpisodicResult (.acknowledged / .id)
# before
doc_id = app.memory.save_episode(title="Quote chat", content=summary, user_id="u1")
if doc_id:
...
# after
episode = app.memory.save_episode(
title="Quote chat",
content=summary,
user_id="u1",
metadata={"channel": "web", "priority": "high"}, # optional caller-supplied metadata
# session_id from ambient context, or pass explicitly
)
if episode.acknowledged:
print(episode.id) # may be "" on app-bound
# search_episodes: dict.get → MemoryChunk metadata + content; pin top_k if needed
# before
episodes = app.memory.search_episodes(query="policy quote", top_k=10)
for ep in episodes:
print(ep.get("title"), ep.get("content"))
# after
episodes = app.memory.search_episodes(query="policy quote", top_k=10)
for ep in episodes:
print(ep.metadata.get("title"), ep.content)
# build_context: str → ContextResponse.formatted_context
# before
prompt = app.memory.build_context(query="help me", user_id="u1")
# after
context = app.memory.build_context(query="help me", user_id="u1")
prompt = context.formatted_context

引用迁移(在助手引擎示例存储库中):

  • agents/insurance-agent/src/insurance_agent/main.py

每个后端的功能差异和应用程序限制的差距记录在内存包功能矩阵中。 Memory 外观的方法级文档位于agent-engine-sdk-memory 中。

使用 agent.yaml 中的 features.memory: true 启用内存,或者当省略该功能标志时使用旧版 ENABLE_MEMORY=true 环境变量启用内存。现有应用仍可将 enable_memory=... 或 enable_tracing=... 传递给 App(...),但这些构造函数标志已弃用:将内存移至 agent.yaml,并完全删除enable_tracing,因为跟踪始终开启。

技能名为 Markdown 文件 (SKILL.md),LLM 可通过渐进式披露按需加载该文件。使用它们对领域专业知识(例如安全查看规则、编码约定)进行编码,如果始终包含这些专业知识,则会导致系统提示臃肿。

my_agent/
skills/
security-checklist/
SKILL.md
style-guide/
SKILL.md

将 skills/ 父目录传递给 skills=[...]。在运行时,deepagents 通过配置的后端列出该目录,并发现每个包含 SKILL.md 作为一项技能的直接目录。发现深一层,而不是递归的。

---
name: security-checklist
description: Security review rules for Python code, focusing on injection and auth
---
# Security Checklist
## REVIEW-RULE-ID-SEC-1: SQL injection
Never concatenate user input into SQL...
## REVIEW-RULE-ID-SEC-2: Command / path injection
Calls to `subprocess.run`, `os.system`, `shell=True`, and `open()` must not
interpolate untrusted input...

Deepagents 在运行时验证技能frontmatter。它会跳过不可读或无法解析的 frontmatter 以及缺少 name 或 description 的技能;客服技能命名或目录名称违规会产生警告,但仍可能加载。 SDK 会转发已声明的路径,而不对其进行检查或筛选。

注意

先决条件: agent.yaml 必须包含 features.deep_agent: true,否则 App.deep_agent() 将在构造时引发 RuntimeError:

features:
deep_agent: true
from langchain_openai import ChatOpenAI
from agent_engine_sdk_langgraph import App
app = App(app_name="My Reviewer")
@app.entrypoint
def build_agent():
return app.deep_agent(
llm=ChatOpenAI(model="gpt-5.4"),
system_prompt="You are a code reviewer.",
skills=["skills"],
)
app.run()

每个 skills=[...] 条目都是父源目录,而不是叶技能目录或 SKILL.md文件。路径是相对于包含 agent.yaml 的目录的,因此 skills=["skills"] 适用于单代理映像 (/app/skills) 和单一存储库映像 (/app/<agent-subdirectory>/skills)。如果您的技能位于代理源代码树内的其他位置,请将 AGENTIC_SKILLS_DIR设立为该相对目录,并使 skills=[...] 相对于该目录。在运行时,deepagents 从每个发现的技能中读取 frontmatter,并将其元数据(名称、描述和解析路径)作为系统提示符中的技能系统区块传递给 LLM。

在第 1 回合,LLM 只能看到技能元数据,而看不到技能体。当用户询问安全性时,LLM 决定 read_file("<agent-dir>/skills/security-checklist/SKILL.md"),并且全文作为 ToolMessage 到达上下文中,轮流 2。

这样可以保持基本提示的简洁性(元数据大约是每个技能的50 个词元),同时允许按需加载深入的专业知识。

ToolPod 的可写文件系统和Shell处理程序仍使用 WORKSPACE_DIR,即默认为 /tmp/agent-workspace。将其保留为暂存空间。

捆绑的技能将被视为只读资源。 ToolPod 从 AGENTIC_AGENT_CONFIG_PATH 或 AGENTIC_AGENT_WORKDIR 派生出默认技能根:如果运行时配置为 /app/agent.yaml,则技能根为 /app/skills;如果运行时配置为 /app/agents/reviewer/agent.yaml,则技能根为 /app/agents/reviewer/skills。 AGENTIC_SKILLS_DIR 会覆盖该根,并且必须相对于代理源根。只读文件系统工具可以加载该根目录下的文件,而无需将 WORKSPACE_DIR 设置为技能目录。写入、编辑和Shell操作仍保留在可写工作区中。技能根在 Tool Pod初创企业解析,而不是在 SDK 导入时解析,因此普通的静态 SDK 导入可以正常工作 — 不需要导入顺序解决方法。

技能仅对声明技能的代理可见。如果您的代理生成了子代理(通过 task 工具),这些子代理不会继承父代理的技能。在每个需要技能文件的子代理规范上传递 skills=[...]。

深度代理运行时为内置工具保留 9 工具名称。请勿使用以下任何名称注册 @app.tool() — 它会默默地隐藏内置函数并破坏技能/沙盒行为:

  • read_file、write_file、edit_file、ls、glob、grep(文件系统)

  • execute (Shell)

  • write_todos (规划中)

  • task (subagent dispatch)

选择发生冲突的名称会默默地影响内置— 不会出现导入时错误。

目标 <每个 SKILL.md 正文 200 行。更大的技能:

  • 加载时使用更多上下文(每个 read_file 都是一个全文转储)

  • 在复杂回合中遇到 LLM 的单消息上下文限制的风险

  • 建议将技能分割为多个有针对性的文件

运行时在线程的生命周期内以代理状态缓存 skills_metadata。如果编辑 SKILL.md文件,现有线程将继续使用过时的元元数据,直到重置。在开发中:删除线程或启动新会话。在生产中:技能更改应与新模型/提示版本的发布同时进行。

有关参考实施,请参阅代理引擎示例存储库中的 Code Reviewer 代理:

  • 助手引擎示例存储库agents/code-reviewer-agent/src/code_reviewer_agent/main.py — 连线

  • 助手引擎示例存储库agents/code-reviewer-agent/skills/*/SKILL.md —示例技能

LangGraphBaseAgent.stream() 生成 StreamEvent 对象。使用 async for 进行迭代以接收词元级更新、子代理生命周期标记和最终结果。

event
当它触发时
data 字段

token

来自根代理或任何活动子代理的每个 LLM 令牌数据数据块。

content:令牌文本。 source:"" 表示根代理,或子代理的图表名称。 tool_call_id:当此子代理处于飞行状态时,父代理的 task 工具调用 ID(在组装之前可能为 "")。

subagent_start

子代理开始运行。从以下两个路径之一发出: (1)主节点 (primary node in the replica set)— 观察到父代理的 task 工具调用; (2) 合成回退 — 有源令牌在父工具调用组装之前到达(缓冲调度提供程序)。这两条路径相互去重,因此每个子代理每轮只触发一次启动。

source /subagent_name:子代理的图表名称。 tool_call_id:父项的 task tool_call_id,如果在组装 tool_call 之前触发了合成回退,则为 ""。 description:父进程传递给 task 的 description 参数(当合成首先触发时为空)。

subagent_end

子代理运行完成。主路径:父图表观察到一个 Command-close,其 tool_call_id 与一个打开的子代理匹配。防御路径:流错误或使用悬空子代理完成 — 每个孤立子代理发出一个 subagent_end,以便使用者可以关闭其用户界面状态。

source /subagent_name:与启动相同。 tool_call_id:结束 tool_call_id,或来自从未收到过真正的tool_call 的仅合成开始的孤立结束的 ""。 summary:子代理的最终 ToolMessage 内容(防守端为空)。

result

最终完成 root代理。

response 加上完整的消息列表。

suspend

HITL 中断 —图表暂停,等待人工查看。

suspend_payload, checkpoint_id.

消费者须知:

  • subagent_start 和 subagent_end 始终成对出现,包括异常终止时。 stream() 中的清理臂将 GeneratorExit(使用方断开连接 — 丢弃状态而不让出,因为没有使用方剩余)与提供者端错误(让出防御性 subagent_end,然后重新加注)。

  • 对于同一子代理类型的并行调度,每次调用都有自己的 tool_call_id。首选 tool_call_id 作为路由令牌,仅当 tool_call_id == "" 时才回退到 source。

  • subagent_end.summary 是子代理的最终响应文本 — 与父代理将视为 task 工具的返回值的字符串相同。

持久工作流程重放 LangGraph 的原生interrupt() 调用,并将其新生成的原生ID 转换为之前记录的 OE 活动位置。有关完整的暂停、重放和 Command(resume=...) 流程及其代码调用站点,请参阅持久性 LangGraph 中断和恢复。

有关自动生成的 App / LangGraph 图面,请参阅 docs/api.md。有关 Memory 门面(方法、返回类型、身份规则),请参阅agent-engine-sdk-memory 及其功能矩阵。

环境变量
默认
说明

MDB_AGENTIC_STORE_DB

mdb_store

用于 LangGraph 检查点的每个项目的MongoDB存储的基本名称(仅限 AER模式)。除非在下面进行覆盖,否则项目范围界定/发现仍然适用。

CHECKPOINT_DB_NAME

(未设置)

设立后的 MongoDBSaver数据库确切名称。跳过项目范围界定和发现。选择使用双运行时共享检查点数据库(在代理AER Pod env/SecretRefs 上设立)。

CHECKPOINTER_SERVER_SELECTION_TIMEOUT

5.0

在检查点IO 失败之前, MongoDB检查点服务器选择的秒数。

CHECKPOINTER_CONNECT_TIMEOUT

5.0

建立MongoDB检查点连接的秒数。

CHECKPOINTER_SOCKET_TIMEOUT

15.0

MongoDB检查指针套接字读取/写入的秒数。

默认下,LangGraph检查点thread_id 是 session_id:workspace_id。代理可以使用 @app.resolve_thread_id(在刷新和恢复时逐字使用的返回值)覆盖此值。自定义密钥对Atlas助手引擎 /query/sessions* 历史记录查找不可见,后者仍仅使用默认的会话/工作区派生密钥。绕过工作区范围的代理还在检查点数据库中拥有冲突隔离性。

LangGraph 时间旅行将修补源 thread_id。 Atlas Agent Engine 会创建一个新会话。原生检查点会话将已完成的检查点复制到新线程上;持久工作流会话从在受保护的暂存中重建的经 OE 验证的状态分支。请参阅会话分叉与 LangGraph 时间旅行。

平台教学文档(什么是持久工作流、原语、启用、限制):持久工作流。共享工具和 LLM 重放模型在持久活动身份中进行了描述。有关完整的序列图、示例和代码调用站点地图,请参阅持久编译的子图和持久深度代理委派。只有通过Atlas Agent Engine 的安全 LLM 和工具包装器路由的效果才能参与持久性记录/重放。对于持久性的工具重放,ToolCall 必须来自安全的 LLM 包装器,并通过 app.get_tools() 返回的工具执行。

持久工作流程不会公开 LangGraph 的动态 `Send < https://docs.langchain.com/oss/python/langgraph/graph-api#send >`__ 扇出。普通平台检查点会拒绝应用程序编写的 Send 写入。由 app.deep_agent() 创建的图是一个不透明的异常,因为 LangChain 在内部使用 Send 来路由 ToolCall;这种私有兼容性不会使 Send 成为受支持的应用程序API。请改用固定图表边、已编译的子图或深度助手任务委托。原生检查点工作流程不受影响。

读取仅限于作用域。会话历史记录将每个Atlas助手引擎 session_id 扩展到仅其工作区范围的复合键;一旦知道工作区范围,就永远不会查询裸露的无作用域键,因为裸键可由共享存储上的每个工作区读取和写入。因此,历史记录端点不再为作用域界定之前写入的传统检查点提供服务;请勿重新添加回退。空作用域仅在显式未限定作用域的运行时(本地开发/测试,无 APP_ID)上才合法。托管 AER 带有 REQUIRE_PROJECT_SCOPED_DB;如果其中缺少 APP_ID,则读取和写入都将无法关闭,而不是信任传输工作区或使用裸密钥。自定义密钥的生产采用者仍应将共享数据库内的检查点密钥唯一性视为由代理拥有。

  • Python >= 3.11

  • uv

uv sync --extra dev

对于 CI 运行的相同检查(lint + format + Pyright + 测试),请使用存储库根目录中的统一运行器:./scripts/test.sh agent-engine-sdk-langgraph。

uv run pytest
uv run pyright
make docs
# Check for lint errors
uv run ruff check src
# Auto-fix lint errors
uv run ruff check --fix src
# Format code
uv run ruff format src
给本页内容打分