39 KiB
DeepAgents 기술 가이드
목차
- 개요 및 아키텍처
- create_deep_agent() API
- Backend 시스템
- Middleware 시스템
- SubAgent 위임 패턴
- Filesystem Backend 활용 흐름
- 실전 활용 패턴
- Sandbox Backend 및 명령 실행
- Filesystem Backend 주의사항 및 Edge Cases
- 보안 모범 사례
1. 개요 및 아키텍처
1.1 DeepAgents란?
DeepAgents는 복잡한 다단계 작업을 처리하는 "Deep Thinking" AI 에이전트를 구축하기 위한 LangChain 기반 프레임워크입니다. 핵심 특징:
- 파일시스템 기반 컨텍스트 관리: 가상 파일시스템을 통한 장기 메모리
- SubAgent 위임: 독립적인 서브에이전트로 작업 병렬 처리
- 자동 컨텍스트 요약: 170K 토큰 초과 시 자동 메시지 요약
- Human-in-the-Loop: 특정 도구 실행 전 사람 승인 요청
1.2 전체 아키텍처
graph TB
subgraph "DeepAgent 아키텍처"
User[사용자 입력]
subgraph "Main Agent"
Model[LLM Model<br/>Claude/GPT]
Middleware[Middleware Stack]
Tools[Tool Executor]
end
subgraph "Middleware Stack"
TDL[TodoListMiddleware<br/>write_todos 도구]
FSM[FilesystemMiddleware<br/>ls, read, write, edit, glob, grep]
SAM[SubAgentMiddleware<br/>task 도구]
SUM[SummarizationMiddleware<br/>컨텍스트 요약]
PCM[PatchToolCallsMiddleware<br/>dangling tool call 처리]
APM[AnthropicPromptCachingMiddleware<br/>프롬프트 캐싱]
end
subgraph "Backend Layer"
State[StateBackend<br/>임시 상태]
FS[FilesystemBackend<br/>로컬 파일시스템]
Store[StoreBackend<br/>영구 저장소]
Comp[CompositeBackend<br/>경로 라우팅]
end
subgraph "SubAgents"
GP[General-Purpose Agent]
Custom[Custom SubAgents]
end
end
User --> Model
Model --> Middleware
Middleware --> TDL
TDL --> FSM
FSM --> SAM
SAM --> SUM
SUM --> APM
APM --> PCM
PCM --> Tools
FSM --> State
FSM --> FS
FSM --> Store
FSM --> Comp
SAM --> GP
SAM --> Custom
2. create_deep_agent() API
2.1 함수 시그니처
def create_deep_agent(
model: str | BaseChatModel | None = None,
tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
*,
system_prompt: str | None = None,
middleware: Sequence[AgentMiddleware] = (),
subagents: list[SubAgent | CompiledSubAgent] | None = None,
response_format: ResponseFormat | None = None,
context_schema: type[Any] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
backend: BackendProtocol | BackendFactory | None = None,
interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
debug: bool = False,
name: str | None = None,
cache: BaseCache | None = None,
) -> CompiledStateGraph
2.2 핵심 파라미터 상세
| 파라미터 | 타입 | 기본값 | 설명 |
|---|---|---|---|
model |
str | BaseChatModel |
Claude Sonnet 4.5 | 사용할 LLM. 문자열("openai:gpt-4o") 또는 BaseChatModel 인스턴스 |
tools |
Sequence[...] |
None | 에이전트에 추가할 커스텀 도구 (내장 도구 외) |
backend |
BackendProtocol | BackendFactory |
StateBackend | 파일 저장소 백엔드 |
subagents |
list[SubAgent | CompiledSubAgent] |
None | 커스텀 서브에이전트 정의 |
interrupt_on |
dict[str, bool | InterruptOnConfig] |
None | HITL 인터럽트 설정 |
checkpointer |
Checkpointer |
None | 대화 상태 영속화 |
store |
BaseStore |
None | 스레드 간 영구 메모리 |
2.3 기본 사용 예시
from deepagents import create_deep_agent
from langchain.chat_models import init_chat_model
# 기본 에이전트 (Claude Sonnet 4.5)
agent = create_deep_agent()
# GPT-4.1 사용
model = init_chat_model(model="openai:gpt-4.1")
agent = create_deep_agent(model=model)
# 커스텀 도구 추가
def my_search(query: str) -> str:
"""웹 검색을 수행합니다."""
return f"검색 결과: {query}"
agent = create_deep_agent(
model=model,
tools=[my_search],
system_prompt="당신은 리서치 전문가입니다."
)
2.4 자동 주입되는 내장 도구
| 도구 | 출처 | 기능 |
|---|---|---|
write_todos |
TodoListMiddleware | 작업 목록 관리 (pending, in_progress, completed) |
ls |
FilesystemMiddleware | 디렉토리 내용 조회 |
read_file |
FilesystemMiddleware | 파일 읽기 (offset/limit 페이지네이션) |
write_file |
FilesystemMiddleware | 새 파일 생성 |
edit_file |
FilesystemMiddleware | 기존 파일 편집 (문자열 치환) |
glob |
FilesystemMiddleware | 패턴으로 파일 검색 |
grep |
FilesystemMiddleware | 파일 내용 검색 |
execute |
FilesystemMiddleware | 쉘 명령 실행 (SandboxBackend 필요) |
task |
SubAgentMiddleware | 서브에이전트 호출 |
3. Backend 시스템
Backend는 DeepAgent의 파일 저장소 추상화 계층입니다.
모든 Backend는 BackendProtocol을 구현합니다.
3.1 BackendProtocol 인터페이스
class BackendProtocol(ABC):
# 파일 목록 조회
def ls_info(self, path: str) -> list[FileInfo]
async def als_info(self, path: str) -> list[FileInfo]
# 파일 읽기 (페이지네이션 지원)
def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> str
async def aread(self, file_path: str, offset: int = 0, limit: int = 2000) -> str
# 파일 쓰기 (새 파일만, 기존 파일은 에러)
def write(self, file_path: str, content: str) -> WriteResult
async def awrite(self, file_path: str, content: str) -> WriteResult
# 파일 편집 (문자열 치환)
def edit(self, file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> EditResult
async def aedit(self, file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> EditResult
# 검색
def grep_raw(self, pattern: str, path: str | None = None, glob: str | None = None) -> list[GrepMatch] | str
def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]
# 벌크 작업
def upload_files(self, files: list[tuple[str, bytes]]) -> list[FileUploadResponse]
def download_files(self, paths: list[str]) -> list[FileDownloadResponse]
3.2 Backend 구현체 비교
graph LR
subgraph "Backend 종류"
State[StateBackend<br/>임시/상태 기반]
FS[FilesystemBackend<br/>로컬 파일시스템]
Store[StoreBackend<br/>LangGraph Store]
Comp[CompositeBackend<br/>라우팅]
end
subgraph "특성"
Ephemeral[임시 저장<br/>스레드 범위]
Persistent[영구 저장<br/>파일 기반]
CrossThread[스레드 간 공유<br/>namespace 기반]
Routing[경로 기반 라우팅<br/>백엔드 조합]
end
State --> Ephemeral
FS --> Persistent
Store --> CrossThread
Comp --> Routing
| Backend | 저장 위치 | 수명 | files_update 반환 | 사용 시나리오 |
|---|---|---|---|---|
| StateBackend | LangGraph State | 대화 스레드 | (상태 업데이트용) | 임시 작업 파일 |
| FilesystemBackend | 로컬 디스크 | 영구 | (이미 저장됨) | 실제 파일 조작 |
| StoreBackend | LangGraph Store | 스레드 간 영구 | (이미 저장됨) | 장기 메모리 |
| CompositeBackend | 라우팅 | 백엔드별 상이 | 백엔드별 상이 | 하이브리드 |
3.3 StateBackend 상세
LangGraph 상태에 파일을 저장하는 임시 백엔드입니다.
class StateBackend(BackendProtocol):
def __init__(self, runtime: ToolRuntime):
self.runtime = runtime
def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> str:
files = self.runtime.state.get("files", {})
file_data = files.get(file_path)
if file_data is None:
return f"Error: File '{file_path}' not found"
return format_read_response(file_data, offset, limit)
def write(self, file_path: str, content: str) -> WriteResult:
files = self.runtime.state.get("files", {})
if file_path in files:
return WriteResult(error=f"Cannot write to {file_path} because it already exists...")
new_file_data = create_file_data(content)
return WriteResult(path=file_path, files_update={file_path: new_file_data})
FileData 구조:
class FileData(TypedDict):
content: list[str] # 파일 내용 (줄 단위)
created_at: str # ISO 8601 타임스탬프
modified_at: str # ISO 8601 타임스탬프
3.4 FilesystemBackend 상세
로컬 파일시스템에 직접 접근하는 백엔드입니다.
class FilesystemBackend(BackendProtocol):
def __init__(
self,
root_dir: str | Path | None = None,
virtual_mode: bool = False, # 가상 경로 모드 (샌드박싱)
max_file_size_mb: int = 10,
):
self.cwd = Path(root_dir).resolve() if root_dir else Path.cwd()
self.virtual_mode = virtual_mode
보안 기능:
virtual_mode=True: 모든 경로가root_dir하위로 제한 (디렉토리 탈출 방지)O_NOFOLLOW플래그: 심볼릭 링크 따라가기 방지- Path traversal 차단:
..,~패턴 거부
검색 최적화:
- 1차: ripgrep (
rg) JSON 출력 사용 - 2차: Python 폴백 (Ripgrep 미설치 시)
3.5 CompositeBackend 상세
경로 접두사 기반으로 여러 백엔드를 라우팅합니다.
class CompositeBackend:
def __init__(
self,
default: BackendProtocol | StateBackend, # 기본 백엔드
routes: dict[str, BackendProtocol], # 접두사 -> 백엔드 매핑
):
self.default = default
self.routes = routes
# 가장 긴 접두사부터 매칭 (longest-prefix matching)
self.sorted_routes = sorted(routes.items(), key=lambda x: len(x[0]), reverse=True)
사용 예시:
from deepagents.backends import StateBackend, StoreBackend, FilesystemBackend, CompositeBackend
backend = CompositeBackend(
default=StateBackend, # / 하위 기본
routes={
"/memories/": StoreBackend, # 장기 메모리 (영구)
"/workspace/": FilesystemBackend(root_dir="/tmp/agent", virtual_mode=True),
}
)
agent = create_deep_agent(backend=backend)
라우팅 동작:
/notes.txt → StateBackend (default)
/memories/facts.txt → StoreBackend
/workspace/code.py → FilesystemBackend
4. Middleware 시스템
Middleware는 에이전트의 모델 호출과 도구 실행을 가로채서 기능을 주입합니다.
4.1 Middleware 적용 순서
create_deep_agent()가 생성하는 미들웨어 스택:
graph TB
subgraph "Middleware Stack (외부 → 내부)"
HITL[8. HumanInTheLoopMiddleware<br/>선택적 - interrupt_on 설정 시]
User[7. User Middleware<br/>사용자 정의]
PCM[6. PatchToolCallsMiddleware<br/>dangling tool call 수리]
APM[5. AnthropicPromptCachingMiddleware<br/>프롬프트 캐싱]
SUM[4. SummarizationMiddleware<br/>컨텍스트 요약]
SAM[3. SubAgentMiddleware<br/>task 도구 + 서브에이전트]
FSM[2. FilesystemMiddleware<br/>파일시스템 도구]
TDL[1. TodoListMiddleware<br/>작업 관리]
end
Request[요청] --> HITL
HITL --> User
User --> PCM
PCM --> APM
APM --> SUM
SUM --> SAM
SAM --> FSM
FSM --> TDL
TDL --> Model[LLM 호출]
4.2 AgentMiddleware 인터페이스
class AgentMiddleware(ABC):
# 상태 스키마 확장
state_schema: type[AgentState] | None = None
# 주입할 도구
tools: list[BaseTool] = []
# 모델 호출 래핑 (시스템 프롬프트 수정, 도구 필터링 등)
def wrap_model_call(self, request: ModelRequest, handler: Callable) -> ModelResponse
async def awrap_model_call(self, request: ModelRequest, handler: Callable) -> ModelResponse
# 도구 호출 래핑 (결과 가로채기, 상태 업데이트 등)
def wrap_tool_call(self, request: ToolCallRequest, handler: Callable) -> ToolMessage | Command
async def awrap_tool_call(self, request: ToolCallRequest, handler: Callable) -> ToolMessage | Command
# 에이전트 실행 전 훅
def before_agent(self, state: AgentState, runtime: Runtime) -> dict | None
4.3 FilesystemMiddleware
파일시스템 도구를 주입하고 대용량 결과를 자동 관리합니다.
class FilesystemMiddleware(AgentMiddleware):
state_schema = FilesystemState # files: dict[str, FileData] 추가
def __init__(
self,
backend: BACKEND_TYPES | None = None,
system_prompt: str | None = None,
custom_tool_descriptions: dict[str, str] | None = None,
tool_token_limit_before_evict: int | None = 20000, # 대용량 결과 기준
)
대용량 결과 자동 처리:
- 도구 결과가
4 × tool_token_limit_before_evict(기본 80,000자) 초과 시 /large_tool_results/{tool_call_id}경로에 자동 저장- 에이전트에게는 처음 10줄 샘플 + 파일 참조 메시지 반환
- 에이전트가
read_file로 페이지네이션 읽기 가능
4.4 SubAgentMiddleware
서브에이전트 생성 및 task 도구를 주입합니다.
class SubAgentMiddleware(AgentMiddleware):
def __init__(
self,
default_model: str | BaseChatModel,
default_tools: Sequence[...] | None = None,
default_middleware: list[AgentMiddleware] | None = None,
default_interrupt_on: dict[...] | None = None,
subagents: list[SubAgent | CompiledSubAgent] | None = None,
system_prompt: str | None = TASK_SYSTEM_PROMPT,
general_purpose_agent: bool = True, # 기본 범용 에이전트 포함
task_description: str | None = None,
)
SubAgent 정의:
class SubAgent(TypedDict):
name: str # 에이전트 식별자
description: str # 메인 에이전트에게 보여줄 설명
system_prompt: str # 서브에이전트 시스템 프롬프트
tools: Sequence[...] # 사용 가능한 도구
model: NotRequired[str | BaseChatModel] # 모델 오버라이드
middleware: NotRequired[list[AgentMiddleware]] # 추가 미들웨어
interrupt_on: NotRequired[dict[...]] # HITL 설정
4.5 SummarizationMiddleware
컨텍스트 오버플로우 방지를 위한 자동 메시지 요약입니다.
SummarizationMiddleware(
model=model,
trigger=("fraction", 0.85), # 컨텍스트의 85% 도달 시 요약
keep=("fraction", 0.10), # 최근 10% 유지
)
트리거 조건:
("fraction", 0.85): 모델 컨텍스트의 85% 도달 시("tokens", 170000): 170K 토큰 도달 시 (모델 정보 없을 때)
4.6 PatchToolCallsMiddleware
"Dangling tool call" 문제를 해결합니다.
문제 상황: AIMessage에 tool_calls가 있지만 대응하는 ToolMessage가 없는 경우
해결 방법: before_agent 훅에서 누락된 ToolMessage를 자동 생성
ToolMessage(
content="Tool call was cancelled or did not complete.",
tool_call_id=dangling_tool_call_id
)
5. SubAgent 위임 패턴
5.1 SubAgent 동작 원리
sequenceDiagram
participant User as 사용자
participant Main as Main Agent
participant Task as task 도구
participant Sub as SubAgent
User->>Main: 복잡한 요청
Main->>Main: 작업 분석
Main->>Task: task(description, subagent_type)
Note over Task: 상태 격리 준비
Task->>Task: messages, todos 제외
Task->>Sub: HumanMessage(description)
Sub->>Sub: 독립적 실행
Sub->>Sub: 도구 호출들...
Sub-->>Task: 최종 결과 메시지
Note over Task: 결과 반환
Task->>Task: ToolMessage로 변환
Task-->>Main: Command(update={...})
Main->>Main: 결과 통합
Main-->>User: 최종 응답
5.2 상태 격리 메커니즘
제외되는 상태 키 (_EXCLUDED_STATE_KEYS):
{"messages", "todos", "structured_response"}
messages: 서브에이전트는 독립적인 대화 컨텍스트를 가짐todos: 작업 목록 충돌 방지structured_response: 스키마 충돌 방지
전달되는 상태: 위 키를 제외한 모든 커스텀 상태 (예: files, 사용자 정의 상태)
5.3 병렬 SubAgent 실행
메인 에이전트가 한 번의 응답에서 여러 task 도구를 호출하면 병렬 실행됩니다:
# LLM이 생성하는 응답
AIMessage(
tool_calls=[
{"name": "task", "args": {"description": "Python 조사", "subagent_type": "general-purpose"}},
{"name": "task", "args": {"description": "JavaScript 조사", "subagent_type": "general-purpose"}},
{"name": "task", "args": {"description": "Rust 조사", "subagent_type": "general-purpose"}},
]
)
각 서브에이전트는:
- 독립적인 상태 복사본으로 실행
- 병렬 처리되어 성능 최적화
- 개별
ToolMessage로 결과 반환
5.4 커스텀 SubAgent 정의
research_agent = {
"name": "researcher",
"description": "심층 리서치가 필요한 질문에 사용",
"system_prompt": """당신은 전문 리서처입니다.
- 최대 5번의 검색 수행
- 출처 명시 필수
- 결과를 구조화된 형식으로 반환""",
"tools": [tavily_search, think_tool],
"model": "openai:gpt-4o", # 선택적 모델 오버라이드
}
agent = create_deep_agent(
model=model,
subagents=[research_agent],
)
6. Filesystem Backend 활용 흐름
6.1 전체 데이터 흐름
flowchart TB
subgraph "에이전트 실행"
LLM[LLM 모델]
TN[Tool Node]
end
subgraph "FilesystemMiddleware"
FSM_wrap[wrap_tool_call]
FSM_tools[도구 함수들<br/>ls, read, write, edit, glob, grep]
end
subgraph "Backend 선택"
Router{CompositeBackend<br/>경로 라우팅}
State[StateBackend]
FS[FilesystemBackend]
Store[StoreBackend]
end
subgraph "저장소"
LGState[(LangGraph State)]
Disk[(로컬 디스크)]
LGStore[(LangGraph Store)]
end
LLM -->|tool_calls| TN
TN --> FSM_wrap
FSM_wrap --> FSM_tools
FSM_tools --> Router
Router -->|"/"| State
Router -->|"/memories/"| Store
Router -->|"/workspace/"| FS
State -->|files_update| LGState
FS -->|직접 I/O| Disk
Store -->|namespace 저장| LGStore
State -.->|읽기| LGState
FS -.->|읽기| Disk
Store -.->|읽기| LGStore
6.2 파일 쓰기 흐름 상세
sequenceDiagram
participant Agent as 에이전트
participant FSM as FilesystemMiddleware
participant Backend as Backend
participant Storage as 저장소
Agent->>FSM: write_file("/report.md", content)
FSM->>FSM: _validate_path() 검증
FSM->>Backend: backend.write("/report.md", content)
alt StateBackend
Backend->>Backend: 기존 파일 확인
Backend->>Backend: create_file_data(content)
Backend-->>FSM: WriteResult(files_update={...})
FSM-->>Agent: Command(update={"files": {...}})
Note over Agent: LangGraph State 업데이트
else FilesystemBackend
Backend->>Storage: os.open(O_CREAT | O_NOFOLLOW)
Backend->>Storage: f.write(content)
Backend-->>FSM: WriteResult(files_update=None)
FSM-->>Agent: ToolMessage("성공")
Note over Storage: 디스크에 직접 저장됨
end
6.3 대용량 결과 처리 흐름
sequenceDiagram
participant Agent as 에이전트
participant FSM as FilesystemMiddleware
participant Tool as 도구 (grep 등)
participant Backend as Backend
Agent->>FSM: grep("pattern", "/")
FSM->>Tool: grep_raw(...)
Tool-->>FSM: 100,000자 결과
FSM->>FSM: len(result) > 80,000? ✓
FSM->>Backend: write("/large_tool_results/...", result)
Backend-->>FSM: WriteResult
FSM-->>Agent: Command(update={<br/> "files": {...},<br/> "messages": [ToolMessage(<br/> "처음 10줄...<br/> 전체 결과: /large_tool_results/..."<br/> )]<br/>})
Note over Agent: 필요시 read_file로 페이지네이션 읽기
6.4 CompositeBackend 라우팅 흐름
flowchart TD
subgraph "CompositeBackend"
Input[파일 경로 입력]
Route{longest-prefix<br/>매칭}
Route -->|"/memories/*"| MemStrip[접두사 제거<br/>"/memories/note.txt" → "/note.txt"]
Route -->|"/workspace/*"| WorkStrip[접두사 제거<br/>"/workspace/code.py" → "/code.py"]
Route -->|"/*" 기타| Default[기본 백엔드<br/>경로 유지]
MemStrip --> Store[StoreBackend]
WorkStrip --> FS[FilesystemBackend]
Default --> State[StateBackend]
Store --> MemResult[결과에 접두사 복원]
FS --> WorkResult[결과에 접두사 복원]
State --> DefaultResult[결과 그대로]
end
Input --> Route
6.5 실제 구현 예시: 리서치 에이전트
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, FilesystemBackend
from langchain.chat_models import init_chat_model
# 백엔드 구성
backend = CompositeBackend(
default=StateBackend, # 팩토리 (런타임에 인스턴스화)
routes={
"/research/": FilesystemBackend(
root_dir="./research_workspace",
virtual_mode=True,
),
}
)
# 커스텀 도구
def tavily_search(query: str, max_results: int = 3) -> str:
"""웹 검색을 수행합니다."""
# Tavily API 호출...
return results
def think_tool(reflection: str) -> str:
"""전략적 사고를 위한 도구입니다."""
return "생각이 기록되었습니다."
# 에이전트 생성
agent = create_deep_agent(
model=init_chat_model("openai:gpt-4o"),
tools=[tavily_search, think_tool],
backend=backend,
system_prompt="""당신은 리서치 전문가입니다.
## 워크플로우
1. 요청 분석 → /research/request.md에 저장
2. 검색 수행 (최대 5회)
3. 결과 종합 → /research/report.md에 저장
4. 검증 및 완료
## 규칙
- 검색 전 항상 think_tool로 전략 수립
- 출처는 반드시 명시
- 결과는 구조화된 마크다운으로 작성
""",
)
# 실행
result = agent.invoke({
"messages": [HumanMessage(content="2024년 AI 에이전트 트렌드 조사해줘")]
})
7. 실전 활용 패턴
7.1 Human-in-the-Loop 설정
agent = create_deep_agent(
model=model,
interrupt_on={
"write_file": True, # 모든 파일 쓰기에 승인 요청
"execute": { # 명령 실행에 상세 설정
"type": "tool",
"action_description": "쉘 명령 실행",
},
"task": False, # 서브에이전트 호출은 자동 허용
}
)
# 스트리밍으로 인터럽트 처리
for event in agent.stream({"messages": [...]}, stream_mode="updates"):
if "interrupt" in event:
# 사용자 승인 요청 UI 표시
approved = get_user_approval(event["interrupt"])
if not approved:
break
7.2 장기 메모리 구성
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
# 체크포인터: 대화 상태 저장
checkpointer = MemorySaver()
# 스토어: 스레드 간 공유 메모리
store = InMemoryStore()
# 백엔드: 메모리 경로 설정
backend = CompositeBackend(
default=StateBackend,
routes={
"/memories/": StoreBackend, # 영구 메모리
}
)
agent = create_deep_agent(
model=model,
checkpointer=checkpointer,
store=store,
backend=backend,
)
# thread_id로 대화 연속성 유지
config = {"configurable": {"thread_id": "user_123"}}
result = agent.invoke({"messages": [...]}, config=config)
7.3 전문화된 SubAgent 팀 구성
# 리서처 에이전트
researcher = {
"name": "researcher",
"description": "심층 웹 리서치가 필요할 때 사용",
"system_prompt": "당신은 리서치 전문가입니다...",
"tools": [tavily_search, think_tool],
}
# 코더 에이전트
coder = {
"name": "coder",
"description": "코드 작성이나 기술 구현이 필요할 때 사용",
"system_prompt": "당신은 시니어 개발자입니다...",
"tools": [], # 파일시스템 도구만 사용
}
# 리뷰어 에이전트
reviewer = {
"name": "reviewer",
"description": "결과물 검토 및 품질 확인이 필요할 때 사용",
"system_prompt": "당신은 QA 전문가입니다...",
"tools": [],
}
agent = create_deep_agent(
model=model,
subagents=[researcher, coder, reviewer],
system_prompt="""당신은 프로젝트 매니저입니다.
복잡한 작업은 적절한 서브에이전트에게 위임하세요:
- 정보 수집 → researcher
- 구현 → coder
- 검증 → reviewer
병렬 실행이 가능한 작업은 동시에 여러 task 도구를 호출하세요.
""",
)
7.4 디버깅 및 모니터링
# LangSmith 트레이싱 활성화
import os
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_..."
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-deepagent"
# 디버그 모드
agent = create_deep_agent(
model=model,
debug=True,
name="ResearchAgent", # 그래프 이름 지정
)
# 그래프 구조 시각화
print(agent.get_graph().draw_mermaid())
부록: 주요 타입 정의
FileInfo
class FileInfo(TypedDict):
path: str # 필수
is_dir: NotRequired[bool]
size: NotRequired[int] # 바이트
modified_at: NotRequired[str] # ISO 8601
WriteResult / EditResult
@dataclass
class WriteResult:
error: str | None = None
path: str | None = None
files_update: dict[str, Any] | None = None # StateBackend용
@dataclass
class EditResult:
error: str | None = None
path: str | None = None
files_update: dict[str, Any] | None = None
occurrences: int | None = None
GrepMatch
class GrepMatch(TypedDict):
path: str # 파일 경로
line: int # 줄 번호 (1부터)
text: str # 매칭된 줄 내용
ExecuteResponse
@dataclass
class ExecuteResponse:
output: str # stdout + stderr
exit_code: int | None = None
truncated: bool = False # 출력 잘림 여부
8. Sandbox Backend 및 명령 실행
DeepAgent는 격리된 환경에서 쉘 명령을 실행할 수 있는 Sandbox 시스템을 제공합니다.
8.1 SandboxBackendProtocol
BackendProtocol을 확장하여 명령 실행 기능을 추가합니다:
class SandboxBackendProtocol(BackendProtocol):
def execute(self, command: str) -> ExecuteResponse:
"""쉘 명령을 실행하고 결과를 반환합니다."""
async def aexecute(self, command: str) -> ExecuteResponse:
"""비동기 버전"""
@property
def id(self) -> str:
"""샌드박스 고유 식별자"""
8.2 BaseSandbox 구현
모든 Sandbox 구현의 기반 클래스로, 파일 작업을 쉘 명령으로 변환합니다:
flowchart LR
subgraph "BaseSandbox"
Read[read_file]
Write[write_file]
Edit[edit_file]
Grep[grep]
Glob[glob]
end
subgraph "명령 변환"
ReadCmd["python3 -c 'read script'"]
WriteCmd["python3 -c 'write script'"]
EditCmd["python3 -c 'edit script'"]
GrepCmd["grep -F pattern"]
GlobCmd["python3 -c 'glob script'"]
end
Read --> ReadCmd
Write --> WriteCmd
Edit --> EditCmd
Grep --> GrepCmd
Glob --> GlobCmd
ReadCmd --> Execute[execute]
WriteCmd --> Execute
EditCmd --> Execute
GrepCmd --> Execute
GlobCmd --> Execute
보안 패턴 - Base64 인코딩:
# 모든 페이로드는 Base64로 인코딩되어 쉘 인젝션 방지
_WRITE_COMMAND_TEMPLATE = """python3 -c "
import base64
content = base64.b64decode('{content_b64}').decode('utf-8')
with open('{file_path}', 'w') as f:
f.write(content)
" 2>&1"""
8.3 Harbor 통합 (Docker 기반)
HarborSandbox는 Docker 컨테이너에서 명령을 실행합니다:
class HarborSandbox(SandboxBackendProtocol):
def __init__(self, environment: BaseEnvironment):
self.environment = environment # Harbor 환경
async def aexecute(self, command: str) -> ExecuteResponse:
result = await self.environment.exec(command)
return ExecuteResponse(
output=self._filter_tty_noise(result.stdout + result.stderr),
exit_code=result.exit_code,
)
TTY 노이즈 필터링:
"cannot set terminal process group (-1)"제거"no job control in this shell"제거"initialize_job_control: Bad file descriptor"제거
8.4 Sandbox 프로바이더
| 프로바이더 | 작업 디렉토리 | 특징 |
|---|---|---|
| Modal | /workspace |
클라우드 함수 기반, 임시 |
| Runloop | /home/user |
클라우드 devbox, ID로 재사용 가능 |
| Daytona | /home/daytona |
클라우드 워크스페이스 |
프로바이더 사용 예시:
from deepagents_cli.integrations.sandbox_factory import create_sandbox
with create_sandbox(
provider="modal",
setup_script_path="./setup.sh", # 초기화 스크립트
) as sandbox:
result = sandbox.execute("python3 --version")
print(result.output)
8.5 Sandbox 선택 가이드
flowchart TD
Start[실행 환경 필요?]
Start -->|예| Isolated{격리 수준?}
Start -->|아니오| FS[FilesystemBackend]
Isolated -->|최대| Docker[HarborSandbox<br/>Docker 컨테이너]
Isolated -->|중간| Cloud[Modal/Runloop<br/>클라우드 샌드박스]
Isolated -->|최소| Local[BaseSandbox 커스텀<br/>로컬 서브프로세스]
Docker --> UseCase1[프로덕션, 민감한 작업]
Cloud --> UseCase2[개발, 테스트]
Local --> UseCase3[로컬 개발, 신뢰할 수 있는 코드]
9. Filesystem Backend 주의사항 및 Edge Cases
9.1 주요 Gotcha 요약
| 이슈 | 심각도 | 영향 | 해결 방법 |
|---|---|---|---|
| Windows에서 Symlink following | 높음 | 보안: 경로 탈출 가능 | virtual_mode=True 사용 |
| Non-UTF-8 인코딩 미지원 | 중간 | 일부 파일 읽기/쓰기 실패 | 사전에 인코딩 변환 |
max_file_size_mb가 grep에만 적용 |
중간 | 대용량 파일 읽기 시 메모리 문제 | 페이지네이션 사용 |
| 권한 오류 무시 | 높음 | 불완전한 디렉토리 목록 | 로그 모니터링 |
| TOCTOU 레이스 컨디션 | 높음 | 동시 쓰기 실패 | 단일 스레드 사용 또는 재시도 로직 |
| 빈 파일 경고 메시지 반환 | 중간 | API 일관성 문제 | 에러 문자열 파싱 |
9.2 경로 검증 상세
FilesystemMiddleware의 _validate_path() 함수:
def _validate_path(path: str, *, allowed_prefixes: Sequence[str] | None = None) -> str:
# 1. Path traversal 차단
if ".." in path or path.startswith("~"):
raise ValueError(f"Path traversal not allowed: {path}")
# 2. Windows 절대 경로 거부
if re.match(r"^[a-zA-Z]:", path): # C:, D:, 등
raise ValueError(f"Windows absolute paths are not supported: {path}")
# 3. 정규화 (// 제거, . 제거)
normalized = os.path.normpath(path).replace("\\", "/")
# 4. 선행 슬래시 보장
if not normalized.startswith("/"):
normalized = f"/{normalized}"
# 5. 허용된 접두사 검증 (선택)
if allowed_prefixes is not None:
if not any(normalized.startswith(prefix) for prefix in allowed_prefixes):
raise ValueError(f"Path must start with one of {allowed_prefixes}")
return normalized
검증 예시:
_validate_path("foo/bar") # → "/foo/bar" ✓
_validate_path("/./foo//bar") # → "/foo/bar" ✓
_validate_path("../etc/passwd") # → ValueError ✗
_validate_path("C:\\Users\\file") # → ValueError ✗
_validate_path("/data/f.txt", allowed_prefixes=["/data/"]) # ✓
_validate_path("/etc/f.txt", allowed_prefixes=["/data/"]) # → ValueError ✗
9.3 Symlink 보안
# FilesystemBackend에서 O_NOFOLLOW 사용
fd = os.open(resolved_path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
플랫폼별 동작:
- Linux/macOS:
O_NOFOLLOW가 symlink following 방지 - Windows:
O_NOFOLLOW미지원, symlink 따라감 ⚠️
권장 사항: Windows 환경에서는 반드시 virtual_mode=True 사용
9.4 인코딩 처리
모든 텍스트 작업은 UTF-8 전용:
# 읽기/쓰기 모두 UTF-8
with os.fdopen(fd, "r", encoding="utf-8") as f:
content = f.read()
Non-UTF-8 파일 처리 시:
# 사전 변환 필요
with open(path, "r", encoding="latin-1") as f:
content = f.read()
utf8_content = content.encode("utf-8").decode("utf-8")
9.5 Edit 작업 주의사항
replace_all 플래그 동작:
flowchart TD
Input[edit_file 호출]
Count[old_string 발생 횟수 카운트]
Input --> Count
Count -->|0회| Error1["에러: String not found"]
Count -->|1회| Replace[치환 수행]
Count -->|2회 이상| Check{replace_all?}
Check -->|True| ReplaceAll[모든 발생 치환]
Check -->|False| Error2["에러: appears N times<br/>Use replace_all=True"]
Replace --> Success[성공]
ReplaceAll --> Success
예시:
# 파일 내용: "hello world hello"
# ❌ 실패 - 2회 발생
result = be.edit("/file.txt", "hello", "hi", replace_all=False)
# → "Error: String 'hello' appears 2 times in file. Use replace_all=True..."
# ✓ 성공
result = be.edit("/file.txt", "hello", "hi", replace_all=True)
# → occurrences=2
9.6 대용량 파일 처리
제한 및 임계값:
| 설정 | 값 | 적용 대상 |
|---|---|---|
max_file_size_mb |
10 MB | grep 검색에서 스킵 |
tool_token_limit_before_evict |
20,000 토큰 | 결과 자동 저장 임계값 |
read() limit |
2,000 줄 | 페이지네이션 기본값 |
| 줄 길이 제한 | 10,000 자 | 포맷팅 시 분할 |
대용량 파일 읽기 패턴:
# 처음 100줄 읽기
content = backend.read("/large_file.txt", offset=0, limit=100)
# 다음 100줄 읽기
content = backend.read("/large_file.txt", offset=100, limit=100)
9.7 동시성 주의
TOCTOU (Time-of-Check-Time-of-Use) 문제:
sequenceDiagram
participant A as 프로세스 A
participant B as 프로세스 B
participant FS as 파일시스템
A->>FS: 파일 존재 확인 (없음)
B->>FS: 파일 존재 확인 (없음)
A->>FS: 파일 생성 시도
B->>FS: 파일 생성 시도
FS-->>A: 성공
FS-->>B: 실패 (이미 존재)
해결 방법:
- 단일 스레드/프로세스로 파일 작업 제한
- 실패 시 재시도 로직 구현
- 외부 잠금 메커니즘 사용
9.8 에러 반환 패턴
두 가지 에러 패턴 혼재:
| 함수 | 반환 타입 | 에러 표현 |
|---|---|---|
read() |
str |
에러 문자열 직접 반환 |
grep_raw() |
list[GrepMatch] | str |
에러 시 문자열 |
write() |
WriteResult |
.error 필드 |
edit() |
EditResult |
.error 필드 |
upload_files() |
list[FileUploadResponse] |
각 항목의 .error 필드 |
안전한 에러 처리:
# read() 결과 처리
result = backend.read("/file.txt")
if result.startswith("Error:"):
handle_error(result)
else:
process_content(result)
# grep_raw() 결과 처리
result = backend.grep_raw("pattern", "/")
if isinstance(result, str): # 에러 문자열
handle_error(result)
else: # list[GrepMatch]
process_matches(result)
10. 보안 모범 사례
10.1 방어 계층
flowchart TB
subgraph "Layer 1: 입력 검증"
PathVal[_validate_path<br/>경로 순회 차단]
SanitizeID[sanitize_tool_call_id<br/>ID 안전화]
end
subgraph "Layer 2: 파일시스템 보호"
Virtual[virtual_mode<br/>루트 격리]
NoFollow[O_NOFOLLOW<br/>symlink 차단]
end
subgraph "Layer 3: 실행 격리"
Base64[Base64 인코딩<br/>인젝션 방지]
Sandbox[Sandbox 실행<br/>프로세스 격리]
end
subgraph "Layer 4: 결과 제한"
TokenLimit[토큰 제한<br/>컨텍스트 보호]
Eviction[대용량 결과 저장<br/>메모리 보호]
end
Input[사용자 입력] --> PathVal
PathVal --> Virtual
Virtual --> Base64
Base64 --> TokenLimit
10.2 보안 체크리스트
# ✓ 프로덕션 설정 예시
agent = create_deep_agent(
model=model,
backend=CompositeBackend(
default=StateBackend, # 임시 파일은 상태에
routes={
"/workspace/": FilesystemBackend(
root_dir="./sandbox",
virtual_mode=True, # ✓ 필수: 경로 격리
max_file_size_mb=5, # ✓ 권장: 파일 크기 제한
),
}
),
middleware=[
FilesystemMiddleware(
tool_token_limit_before_evict=10000, # ✓ 권장: 결과 크기 제한
)
],
interrupt_on={
"execute": True, # ✓ 필수: 명령 실행 전 승인
"write_file": True, # ✓ 권장: 파일 쓰기 전 승인
},
)
10.3 민감한 경로 보호
# allowed_prefixes로 접근 제한
def create_restricted_backend(runtime):
return FilesystemBackend(
root_dir="./data",
virtual_mode=True,
)
# 미들웨어에서 경로 검증
class RestrictedFilesystemMiddleware(FilesystemMiddleware):
def _validate_path(self, path):
# 추가 검증 로직
if "/secrets/" in path or "/.env" in path:
raise ValueError("Access to sensitive paths is forbidden")
return super()._validate_path(path)