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") 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``: 베어
bool/dict가 아닌acknowledged를 사용하여 반환 유형 결과(CreateSemanticResult,CreateEpisodicResult등)를 작성합니다.if result.acknowledged:를 선호합니다(if result:가 아님 — Pydantic 모델은 항상 진실됨). 유사성 검색은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를 전달하세요.ID: 확인할 수 없는 필수 필드는
MemoryIdentityError를 발생시킵니다(더 이상 소프트None/자동 건너뛰기 없음).save_episode에는 해석 가능한session_id이(가) 필요합니다(주변 호출 컨텍스트도 괜찮습니다. 그렇지 않으면 명시적으로 전달하거나MemoryRequestContext를 바인딩합니다). 빈 값은 설정하다 것으로 계산되지 않습니다.앱 바운드 생성 충실도: create-result
id는""일 수 있고has_embedding는 일반적으로False—id가 아닌.acknowledged의 게이트 성공 입니다. 작업별 세부 정보는 메모리 패키지 역량 매트릭스에 나와 있습니다.가져오기 대 검색:
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
참조마이그레이션 ( Agent Engine 예제 리포지토리 내):
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을 완전히 제거 .
기술
스킬은 LLM이 점진적 공개를 통해 온디맨드 방식으로 로드할 수 있는 명명된 마크다운 파일(SKILL.md)입니다. 항상 포함할 경우 시스템 프롬프트가 부풀어 오르는 도메인 전문성 (예: 보안 검토 규칙, 코딩 규칙)을 인코딩하는 데 사용합니다.
디렉토리 레이아웃
my_agent/ skills/ security-checklist/ SKILL.md style-guide/ SKILL.md
상위 skills/ 디렉토리 skills=[...]에 전달합니다. 런타임에 딥에이전트는 구성된 백엔드 통해 해당 디렉토리 나열하고 SKILL.md 를 하나의 스킬 로 포함하는 각 직계 하위 디렉토리 검색합니다. 검색은 재귀적이지 않은 수준으로 한 단계 진행됩니다.
스킬.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...
딥에이전트는 런타임에 스킬 프론트매터의 유효성을 검사합니다. 읽을 수 없거나 구문 분석할 수 없는 전면 내용과 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)와 단일 리포지토리 이미지(/app/<agent-subdirectory>/skills) 모두에 작동합니다. 스킬이 에이전트 소스 트리 내부의 다른 곳에 있는 경우 AGENTIC_SKILLS_DIR를 해당 상대 디렉토리 로 설정하다 하고 skills=[...]을 이를 상대 디렉토리로 설정합니다. 런타임에 딥에이전트는 검색된 각 스킬 에서 프론트매터를 읽고 해당 스킬의 메타데이터 (이름, 설명 및 확인된 경로)를 시스템 프롬프트의 스킬 시스템 차단 으로 LLM에 전달합니다.
점진적 공개
1 턴에서 LLM은 본문이 아닌 스킬 메타데이터 만 볼 수 있습니다. 사용자가 보안에 대해 질문하면 LLM은 read_file("<agent-dir>/skills/security-checklist/SKILL.md")을 결정하고, 2 차례의 컨텍스트에서 전신이 ToolMessage로 도착합니다.
이렇게 하면 기본 프롬프트를 간결하게 유지하면서(메타데이터 는 스킬 당 대략 50 토큰) 필요에 따라 심층적인 전문성 을 로드할 수 있습니다.
번들로 제공되는 스킬 파일 및 샌드박싱
ToolPod의 쓰기 가능한 파일 시스템 및 셸 핸들러는 여전히 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 을 기술 디렉토리 로 설정하지 않고도 해당 루트 아래에 파일을 로드할 수 있습니다. 쓰기, 편집 및 셸 작업은 여전히 쓰기 가능한 작업 공간에서 유지됩니다. 스킬 루트는 SDK 가져오기 시점이 아닌 도구 파드 스타트업 시 확인되므로 일반적인 정적 SDK 가져오기가 작동하며 가져오기 순서 해결 방법이 필요하지 않습니다.
하위 에이전트 비상속
스킬은 해당 스킬을 선언한 에이전트 만 볼 수 있습니다. 에이전트 (
task도구를 통해) 하위 에이전트를 생성하는 경우, 해당 하위 에이전트는 상위 에이전트의 스킬을 상속하지 않습니다. 스킬 파일이 필요한 각 하위 에이전트 사양에 대해skills=[...]을 전달합니다.
예약된 도구 이름
딥 에이전트 런타임은 내장된 도구에 대해 9 도구 이름을 예약합니다. 다음과 같은 이름으로 @app.tool()을 ( 를) 등록하지 마세요.
read_file,write_file,edit_file,ls,glob,grep(파일 시스템)execute(셸)write_todos(planning)task(subagent dispatch)
충돌하는 이름을 선택하면 내장 이름이 자동으로 섀도잉 처리되므로 가져오기 시간 오류는 발생하지 않습니다.
사이즈 지침
대상 < SKILL.md 본문당 200줄. 대규모 기술:
로드 시 더 많은 컨텍스트 소비(각
read_file은(는) 전신 덤프입니다)복잡한 턴에서 LLM의 단일 메시지 컨텍스트 제한에 도달할 위험
스킬 여러 개의 포커스 파일로 분할 해야 한다고 제안합니다.
팁: SKILL.md 편집 후 스레드 재설정
런타임은 스레드의 수명 동안 에이전트 상태 에서 skills_metadata를 캐시합니다. SKILL.md 파일 편집하면 기존 스레드는 재설정될 때까지 오래된 메타데이터 계속 사용합니다. In dev: 스레드를 삭제 하거나 새 세션을 시작합니다. 프로덕션 환경: 스킬 변경은 새 모델/프롬프트 버전 롤아웃과 함께 이루어져야 합니다.
전체 최소 예시
참조 구현 Agent Engine 예제 리포지토리 의 Code Reviewer Agent를 참조하세요.
Agent Engine 예제 리포지토리
agents/code-reviewer-agent/src/code_reviewer_agent/main.py— 연결Agent Engine 예시 리포지토리
agents/code-reviewer-agent/skills/*/SKILL.md— 예시 기술
스트리밍
LangGraphBaseAgent.stream() StreamEvent 객체를 생성합니다. async for로 반복하여 토큰 수준 업데이트, 하위 에이전트 수명 주기 마커 및 최종 결과를 수신합니다.
event | 실행 시 | data 필드 |
|---|---|---|
| 루트 에이전트 또는 활성 하위 에이전트의 각 LLM 토큰 청크 . |
|
| 하위 에이전트 실행 시작됩니다. 두 경로 중 하나에서 방출: (1) 프라이머리 — 상위 에이전트의 |
|
| 하위 에이전트 실행 완료됩니다. 프라이머리 경로: 상위 그래프 |
|
| 루트 에이전트 의 최종 완료 . |
|
| HITL 인터럽트 - 그래프 일시 중지되어 사람의 검토 기다리고 있습니다. |
|
소비자 참고 사항:
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=...) 흐름과 해당 코드 호출 사이트에 대해서는 Durable LangGraph 중단 및 재개를 참조하세요.
API 참조
자동 생성된 앱/LangGraph 표면에 대해서는 docs/api.md를 참조하세요. Memory 파사드(메서드, 반환 유형, ID 규칙)는 Agent-engine-sdk-memory 및 해당 역량 매트릭스를 참조하세요.
구성
환경 변수 | 기본값 | 설명 |
|---|---|---|
|
| LangGraph 체크포인트에 사용되는 프로젝트별 MongoDB 저장 의 기본 이름입니다(AER 모드 에만 해당). 프로젝트 범위/검색은 아래에서 재정의하지 않는 한 계속 적용됩니다. |
| (unset) | 설정하다 시 정확한 MongoDBSaver 데이터베이스 이름입니다. 프로젝트 범위 지정 및 검색을 건너뜁니다. 이중 런타임 공유 체크포인트 DB를 옵트인합니다( 에이전트 AER pod 환경/SecretRefs에서 설정하다 ). |
|
| 체크포인트 IO가 실패하기 전에 MongoDB 체크포인트 서버 선택에 걸리는 시간(초)입니다. |
|
| MongoDB 체크포인터 연결 설정에 걸리는 시간(초)입니다. |
|
| MongoDB 체크포인터 소켓 읽기/쓰기에 소요되는 시간(초)입니다. |
기본값 으로 LangGraph 체크포인트 thread_id는 session_id:workspace_id입니다. 상담원은 @app.resolve_thread_id로 이를 재정의할 수 있습니다(새로고 및 재개할 때 그대로 사용되는 반환 값). 사용자 지정 키는 여전히 기본값 세션/작업 공간에서 파생된 키만 사용하는 Atlas Agent Engine /query/sessions* 기록 조회에는 표시되지 않습니다. 작업 공간 범위를 우회하는 에이전트는 체크포인트 데이터베이스 내에서 충돌 격리 도 소유합니다.
LangGraph 시간 이동은 소스 thread_id를 패치합니다. Atlas Agent Engine은 대신 새 세션을 생성합니다. 네이티브 체크포인트 세션은 완료된 체크포인트 새 스레드에 복사합니다. 지속형 워크플로 세션은 OE 검증 상태 에서 분리된 스크래치로 재구성됩니다.세션 포크 대 LangGraph 시간 이동을 참조하세요.
플랫폼 교육 문서(Durable Workflow의 정의, 기본, 지원, 제한 사항): Durable Workflow. 공유된 도구 및 LLM 재생 모델은 지속형 활동 ID에 설명되어 있습니다. 전체 시퀀스 다이어그램, 예제 및 코드 호출 사이트 맵은 Durable 컴파일된 하위 그래프 및 Durable 딥 에이전트 위임을 참조하세요. Atlas Agent Engine의 보안 LLM 및 도구 래퍼를 통해 라우팅된 효과만 지속형 기록/재생에 참여합니다. 지속형 도구 재생의 경우, 도구 호출은 보안 LLM 래퍼에서 가져와야 하며 app.get_tools()이(가) 반환한 도구를 통해 실행되어야 합니다.
내구성 있는 워크플로는 LangGraph의 동적 `Send <https://docs.langchain.com/oss/python/langgraph/graph-api#send>`__ 팬아웃. 일반 플랫폼 체크포인트는 애플리케이션 작성 Send 쓰기를 거부합니다. LangChain이 내부적으로 Send을(를) 사용하여 도구 호출을 라우팅하기 때문에 app.deep_agent()에 의해 생성된 그래프는 불투명한 예외입니다. 이 비공개 호환성은 Send를 지원되는 애플리케이션 API 만들지 않습니다. 대신 고정 그래프 에지, 컴파일된 하위 그래프 또는 딥 에이전트 작업 위임을 사용합니다. 네이티브 체크포인트 워크플로는 영향을 받지 않습니다.
읽기는 범위 지정 전용입니다. 세션 기록은 각 Atlas Agent Engine session_id을 작업 공간 범위의 복합 키로 확장합니다. 베어 키는 공유 저장 의 모든 작업 공간에서 읽고 쓸 수 있기 때문에 작업 공간 범위가 알려지면 범위가 지정되지 않은 베어 키는 쿼리되지 않습니다. 따라서 범위 지정이 존재하기 전에 작성된 레거시 체크포인트는 더 이상 기록 엔드포인트에서 제공되지 않습니다. 대체를 다시 추가하지 마세요. 빈 범위는 명시적으로 범위가 지정되지 않은 런타임(로컬 개발/테스트, APP_ID 없음)에서만 합법적입니다. 관리되는 AER은 REQUIRE_PROJECT_SCOPED_DB를 전달합니다. APP_ID가 누락된 경우, 유선 작업 공간을 신뢰하거나 베어 키를 사용하는 대신 읽기 및 쓰기가 모두 페일클로즈됩니다. 사용자 지정 키를 프로덕션에 도입하는 경우에도 공유 DB 내부의 체크포인트 키 고유성을 에이전트 소유로 취급해야 합니다.
개발
요구 사항:
Python >= 3.11
개발자 설정
uv sync --extra dev
테스트
CI가 실행하는 동일한 검사(lint + format + pyright + 테스트)의 경우, 저장소 루트에서 통합 러너 ./scripts/test.sh agent-engine-sdk-langgraph를 사용합니다.
uv run pytest
유형 확인
uv run pyright
API Docs 재생성
make docs
Linting & 서식 지정
# 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