MongoDB Atlasエージェント エンジン用の Lgachein SDK。 agent-engine-runner-shared プラットフォームのランタイム上で MongoDB Atlas に固有の軽量ラッパーを提供します。
クイック スタート
インストール
pip install agent-engine-sdk-langgraph
または ucプロジェクトの場合は次のようになります。
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") def lookup(query: str) -> str: """Search the knowledge base.""" return "result for " + query 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 ファサード(プラットフォーム ランタイムを超えるアプリバウンド)です。 ID は、 引数、バインドされたコンテキスト、または環境実行コンテキストからの呼び出しごとに解決されます。
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)は残ります。プラットフォームはテナンシーをスタンプし、永続的な実行を通じて応答を返します。 session_id は、stm ソースが要求された場合にのみ必要です。
from agent_engine_sdk_langgraph import Memory は agent_engine_sdk_memory.Memory と同じクラスを再エクスポートします。 Memory(api_key=...) / Memory(base_url=...) を自分で構築するのはHTTP/ Direct パスであり、アプリバウンドではありません。
これが壊れたとき
app.memoryこれはハードウェアです。 にはデュアルAPIと互換性シティーはありません。カスタマー エージェントは、この SDK を含むドライバーベースのファイルを選択するエージェントイメージを再構築した場合にのみ中断されます。プラットフォームマージのみを使用しても、古いエージェントイメージを再配置しても、そのイメージにすでに埋め込まれている SDK コードは変更されません。保存済みメモリ データの移行はありません。クライアントは戻り値の形状と呼び出し規則の変更のみを返します。
変更前と照合
型 / 真実/メタデータ/``Build_context``
CreateSemanticResultCreateEpisodicResultacknowledgedboolif result.acknowledged:if result:list[MemoryChunk]を返します:chunk.contentchunk.similarity_score書込み (write)titleでは、 とともに型指定された結果( 、 など)が返されます ( /decimals は必須ではありません)。 を優先します(summarytagstermdefinitionではありません -related_termsPrivate モデルは常に true です)。類似性検索では ( 、任意の )が返されます。最上位の辞書キー( 、 、 、 、 、 、...)で使用されていたフィールドは、chunk.metadataの下に存在します(例:ep.metadata.get("title")ep.get("title"))。build_contextは を返します。戻り値をContextResponsecontext.formatted_contextstring としてではなく、 を使用します。書込みヘルパーはキーワード専用(
save_semantic(text=..., label=..., …))です。位置呼び出しではTypeErrorが発生します。``visibility=================================================================================
visibilityuser_id``top_k``:
search_semanticsearch_episodessearch_taxonomictop_k=5010discover_procedures10Memory.search()top_k=10build_contexttop_kmax_tokens500ソースごとの / / のデフォルトは (多くの場合 )でした。 のデフォルトは引き続き です。 と統合してもデフォルトは に設定されています。パブリック には パラメータはありません。グローバル コンテキスト構築予算(取得コストではない)には を使用します。検索とランキングの後、サーバーは トークンのフォーマット予約を減算し、残りに収まるメモリ500 チャンク全体を必要に応じて選択します。 以下の正の値では、メモリの予算はありません。500 を超える値では、チャンクが収まらない場合でも空のコンテキストが生成されることがあります。古い制限が必要な場合は、検索ヘルパーに明示的なtop_kを渡します。ID:
MemoryIdentityErrorNonesave_episodesession_idMemoryRequestContext解決できない必須フィールドでは、 が発生します(ソフト / サイレント スキップはなくなりました)。 には解決可能な が必要です(あいまいな呼び出しコンテキストは正常に行われます。それ以外の場合は、明示的に渡すか、 をバインドします)。空白値は設定されたものとしてカウントされません。アプリバウンド作成の忠実度:
id""has_embeddingFalse作成結果 は.acknowledgedになる可能性があり、 は通常 ( ではなくidで成功するようにします。操作ごとの詳細は、 メモリパッケージ機能マトリックス を参照してください。で検索を取得します。
get_semantic/get_taxonomic_term/list_episodesは引き続き、アプリバウンドで緩やかに型指定された命令を返します。search*メソッドのみがlist[MemoryChunk]を返します。名前を変更します。以前のファケイド名が呼び出された場合は、ファケイド パブリックAPIを使用します(
create_taxonomicsave_taxonomiclist_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 を完全に削除します。
スキル
スキームはマークダウン ファイル(SKILL.md)と呼ばれ、LDM はプログレス 公開経由でオンデマンドで読み込むことができます。これらを使用して、常に含まれているとシステムプロンプトを肥大化させるドメインの専門知識(セキュリティ レビュー ルール、コーディング規則など)をエンコードします。
ディレクトリ レイアウト
my_agent/ skills/ security-checklist/ SKILL.md style-guide/ SKILL.md
親 skills/ディレクトリをskills=[...] に渡します。実行時に、深層は構成されたバックエンドからそのディレクトリを一覧表示し、SKILL.md を 1 つの特権として含む直列の各子ディレクトリを検出します。検出は単一レベルであり、再帰的ではありません。
スキップ
--- 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...
深層は実行時に特権を検証します。読み取れないまたは解析できないフロントマことと、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") 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)と MongoDB イメージ(/app/<agent-subdirectory>/skills)の両方で機能します。スキルがエージェントソース ツリー内の他の場所に存在する場合は、AGENTIC_SKILLS_DIR をその相対ディレクトリに設定し、skills=[...] をそれに相対的にするようにします。実行時に、リージョンは検出された各特権からフロントマを読み取り、そのメタデータ(名前、説明、解決されたパス)をシステム プロンプトの特権システム ブロックとして LM に渡します。
段階的な説明
1をオンにすると、LM にはボディではなくスキップメタデータのみが表示されます。ユーザーがセキュリティについて質問した場合、LMread_file("<agent-dir>/skills/security-checklist/SKILL.md") は と決定し、フルコンテキストはToolMessage 2のコンテキストで として到達します。
これにより、基本的なプロンプトがリージョンに保たれ(メタデータは特権あたり約 50 トークン)、深い専門知識をオンデマンドで読み込むことができます。
バンドルされたツールファイルとサンドボックス
ToolsPd の書込み可能なファイルシステムとシェルハンドラーは引き続き WORKSPACE_DIR を使用し、デフォルトは /tmp/agent-workspace になります。それをスニペット スペースとして保持します。
バンドルされた特権は、代わりに読み取り専用のリソースとして扱われます。 ToolsPd は、デフォルトのスキー ルートを 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 をスキームディレクトリに設定せずに、そのルートの下のファイルをロードできます。書込み、編集、 シェル操作は引き続き書込み可能なワークスペースに残ります。スキーム ルートは、SDK インポート時ではなく、 ツール ポッドのスタートアップ時に解決される ため、通常の静的 SDK インポートが機能します。インポート順序の回避策は必要ありません。
サブエージェントの非継承
特権は、それを宣言するエージェントにのみ表示されます。エージェントがサブエージェント(
taskツールを使用)を生成する場合、それらのサブエージェントは親の操作を継承しません。ルールskills=[...]ファイルを必要とする各サブエージェント仕様に を渡します。
予約されたツール名
ディープエージェント ランタイムでは、組み込み用の9 ツール名が予約されます。これらの名前のいずれかでは、 を登録しないでください。組み込みの が暗黙的にシャドウされ、スキーやサンドボックスの動作が中断されます。@app.tool()
read_file、write_file、edit_file、ls、glob、grep(ファイルシステム)execute(シェル)write_todos(計画)task(subagent dispatch)
競合する名前を選択すると、組み込みが暗黙的にシャドウされます。インポート時のエラーはありません。
サイズガイダンス
ターゲット <200 SR.md ボディあたり 行。より大きな特権:
ロード時により多くのコンテキストを消費するようになりました(各
read_fileはフルボディ ダンプです)複雑な展開で LM の単一メッセージ コンテキスト制限に達するリスク
スキームを複数のフォーカス ファイルに分裂ことをお勧めします
ヒント: EKILL.md 編集後にスレッドをリセットする
ランタイムは、スレッドの有効期間にわたってエージェント状態で skills_metadata をキャッシュします。 SKILL.mdファイルを編集する場合、既存のスレッドはリセットされるまで古いメタデータを使用し続けます。開発環境では、スレッドを削除するか、新しいセッションを開始します。本番環境では: スキーリングの変更は、新しいモデル/プロンプトのバージョンのロールアウトと組み合わせる必要があります。
完全最小の例
参照実装については、「 エージェント エンジンの例のリポジトリ 」で「 コード レビュー エージェント 」を参照してください。
エージェント エンジンの例のリポジトリ
agents/code-reviewer-agent/src/code_reviewer_agent/main.py— ワイヤリングエージェント エンジンの例のリポジトリ
agents/code-reviewer-agent/skills/*/SKILL.md(例特権)
ストリーミング
LangGraphBaseAgent.stream() は StreamEvent オブジェクトを生成します。 async for で反復処理すると、トークンレベルの更新、サブエージェントのライフサイクル マーク、最終結果が得られます。
event | 起動時に | data フィールド |
|---|---|---|
| ルートエージェントまたはアクティブなサブエージェントからの各 LVM トークン チャンク。 |
|
| サブエージェントの実行が開始されます。 2 つのパスのいずれかから発行されます: ()プライマリ(親エージェントの1 |
|
| サブエージェントの実行が終了します。プライマリ パス: 親グラフは、 |
|
| ルートエージェントの最終完了。 |
|
| HITL 割り込み -グラフは人間によるレビューを待機している間に一時停止されます。 |
|
消費者への注釈:
subagent_startとsubagent_endは常にペアになっており、異常終了の場合は を含みます。 のクリーンアップ ARM は、stream()GeneratorExit(コンシューマーの切断 - コンシューマーが存在しないため、中断せずに状態をドロップ)とプロバイダー側のエラー(subagent_endによるディレクティブを返し、その後再発生)とを区別します。同じサブエージェント タイプの並列ディスパッチの場合、各呼び出しには独自の
tool_call_idが含まれます。トークンをルーティングするためにtool_call_idを優先し、tool_call_id == ""の場合にのみsourceにフォールバックします。subagent_end.summaryは、サブエージェントの最終応答テキストです。これは、親エージェントがtaskツールの戻り値とさせるのと同じ文字列です。
耐久性のあるネイティブ割り込み
耐久性のあるワークフローは、LingGraph のネイティブinterrupt() 呼び出しを再生し、新しく生成されたネイティブ ID を以前に記録された OA アクティビティの位置に変換します。完全に停止、リプレイ、Command(resume=...) フローとそのコード呼び出しサイトについては、「 耐久性のある LingGraph の割り込みと再開 」を参照してください。
API リファレンス
自動生成された App / LgGraph 画面については、 docs/api.md を参照してください。Memory ファサード(メソッド、戻り値の型、ID ルール)については、 Agent- engine-sdk-memory とその機能マトリックス を参照してください。
構成
環境変数 | default | 説明 |
|---|---|---|
|
| LgGraph チェックポイントに使用されるプロジェクトごとのMongoDBストアのベース名(AERモードのみ)。プロジェクトのスコープ設定/検出は、以下のようにオーバーライドされない限り、引き続き適用されます。 |
| (設定されていない) | 設定されている場合は、正確な MongoDBSaverデータベース名。プロジェクトのスコープ設定と検出をスキップします。二重実行共有チェックポイントDB のオプトイン(エージェントAER pod env / SecretRefs のエージェントに設定)。 |
|
| チェックポイントIO が失敗する前に、 MongoDBチェックポイントサーバーを選択するための秒数。 |
|
| MongoDBチェックポイント接続を確立するための秒数。 |
|
| MongoDBチェックポイント ソケットの読み取り/書き込みの秒数。 |
デフォルトでは、LingGraphチェックポイントthread_id は session_id:workspace_id です。エージェントは @app.resolve_thread_id (新規および再開時に冗長に使用される戻り値)を使用してこれを上書きできます。カスタムキーは、デフォルトのセッションまたはワークスペースから生成されたキーのみを使用する Atlas Agent Engine /query/sessions* 履歴検索では表示されません。ワークスペースのスコープ設定をバイパスするエージェントも、チェックポイントデータベース内の 衝突分離を備えています。
LongGraph の時間移動がソースthread_id にパッチを適用する。 Atlas Agent Engine は代わりに新しいセッションを作成します。ネイティブ チェックポイント セッションは完了したチェックポイントを新しいスレッドにコピーします。耐久性があるワークフロー セッションは、接続されたスラッシュで再構築された OA 検証済み状態からのブランチ。 「 セッション フォークと LingGraph の時間移動 」を参照してください。
プラットフォーム学習ドキュメント(耐久性のあるワークフローの仕組み、プリミティブ、有効化、制限事項): 耐久性のあるワークフロー。共有ツールと LM リプレイ モデルについて詳しくは、「 耐久性のあるアクティビティ ID 」を参照してください。完全なシーケンス図、例、コード呼び出しサイト マップについては、「 耐久性があるコンパイルされたサブグラフ 」と「 耐久性のあるディープエージェントの削除 」を参照してください。 Atlas Agent Engine の安全な LM およびツール ラッパーを介してルーティングされた影響のみが永続的なレコード/リプレイに参加します。永続的なツールリプレイでは、ツール呼び出しは安全な LMラッパーから取得され、app.get_tools() によって返されたツールを通じて実行される必要があります。
耐久性のあるワークフローでは、LingGraph の動的`Send < が公開されません。 https://docs.lang.com/os/python/langgraph/graph-api#send >の前に試行する必要があります。通常のプラットフォーム チェックポイントは、アプリケーションが承認したSend 書き込みを拒否します。app.deep_agent() Sendによって作成されたグラフは、ツール呼び出しをルーティングするために内部的に を使用するため、不確定の例外です。このプライベート互換性では、Send はサポート対象のアプリケーションAPIではありません。代わりに、固定グラフエッジ、コンパイルされたサブグラフ、または ディープ エージェントのタスクの削除 を使用してください。ネイティブ チェックポイント ワークフローは影響を受けません。
読み取りはスコープ指定のみです。セッション履歴では、各 Atlas エージェント エンジン がワークスペースにスコープが設定された複合キーのみに拡張されます。必要最低限のキーは、共有ストア上のすべてのワークスペースで読み取りと書込みが可能であるため、ワークスペースのスコープが認識されるとクエリされません。スコープ設定が存在する前に書き込まれたレガシー チェックポイントは、履歴エンドポイントでは提供されなくなりました。フォールバックを再追加しないでください。空のスコープは、明示的にスコープが設定されていないランタイム(ローカルsession_id および テスト、APP_ID なし)でのみ正規化されます。マネージド AER には が含まれます。 に欠落している場合は、ワイヤワークスペースを信頼したり、ベアキーを使用したりする代わりに、読み取りと書込みの両方が閉じられません。カスタムREQUIRE_PROJECT_SCOPED_DB APP_IDキーを本番環境で使用する場合でも、共有 DB 内のチェックポイントキーの一意性をエージェント所有ものとして扱う必要があります。
開発
要件
Python >:=3.11
開発者設定
uv sync --extra dev
テスト
CI が実行する同じチェック(lint +形式+ pyrite + テスト)には、統合実行者: リポジトリ ルートからの ./scripts/test.sh agent-engine-sdk-langgraph を使用します。
uv run pytest
型チェック
uv run pyright
APIドキュメントの再生成
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