yanis/deepagent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

DeepAgents with Context Engineering

Browse Source
This commit is contained in:
HyunjunJeon
2026-01-11 18:16:20 +09:00
parent af5fbfabec
commit 6f2c612770
10 changed files with 50 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
Context_Engineering.md
View File

@@ -184,7 +184,7 @@ Manus PDF가 제시하는 3단계 추상화(계층):
### 결론(요약)
- **5대 전략(Offload/Reduce/Retrieve/Isolate/Cache)을 설명하는 데는 충분히 좋습니다.**
- 최근 업데이트로 **“왜 문제인지”, 실패 모드, Tool Offloading** 개요가 추가되어 설명력은 좋아졌지만, 이를 뒷받침하는 **재현 실험/운영 규칙**은 여전히 보완이 필요합니다.
- 특히 이번 업데이트로 **실패 모드(Confusion/Clash/Distraction/Poisoning)를 “실제 실행 + 로그”로 재현하고, 미들웨어로 완화하는 흐름**이 들어가면서 “설명서”로서의 설득력이 크게 좋아졌습니다.
### 잘 표현하는 부분
@@ -194,18 +194,9 @@ Manus PDF가 제시하는 3단계 추상화(계층):
- “구현(DeepAgents 미들웨어/설정)”
- “간단한 시뮬레이션 실험”
로 연결해, “개념 → 구현” 흐름이 명확합니다.
- (추가됨) **실패 모드별 “실제 실행/로그 기반” 미니 실험**이 포함되어, “왜 필요한가 → 어떤 문제가 생기는가 → 어떤 가드레일로 줄이는가”가 한 눈에 연결됩니다.
### 부족한 부분(이 문서 대비 갭)
- **실패 모드 4종(Poisoning/Distraction/Confusion/Clash)**을 “설명”하긴 하지만, 노트북의 **실험/설계에 직접 반영**(재현→완화 비교)되어 있지 않습니다.
- Manus PDF의 중요한 관점인 **“도구도 컨텍스트를 더럽힌다”**(Tool Offloading)도 개요는 추가되었지만, **도구 과다/유사 도구로 인한 혼란**을 줄이는 실험이 없습니다.
- (남은 갭) “Reduce(요약/트리밍)”를 **LangChain `SummarizationMiddleware`의 실제 동작**(토큰 트리거/요약 결과/정보 손실)까지 포함해 재현하려면, 실제 LLM 호출(키/모델)과 더 긴 히스토리가 필요합니다.
- Manus PDF의 메시지인 **“과잉 컨텍스트/과잉 설계 경계(‘Removing, not adding’)”**가 체크리스트/의사결정 가이드로 정리되어 있지 않습니다.
### “Context Engineering 설명”을 완성도 높게 만들기 위한 보완 제안(노트북 기준)
노트북을 “5가지 전략 데모”에서 “Context Engineering 설명서”로 끌어올리려면, 아래 4가지만 추가해도 체감이 큽니다.
1. (완료) **서론 강화(문제 정의)**: 컨텍스트 성장, context rot, 실패 모드 4종을 첫 섹션에서 명시.
2. (추가됨 - 시뮬레이션) **실패 모드별 미니 실험**: Confusion(도구 과다/유사 도구)·Clash(모순 tool output)·Distraction(장기 로그에서 반복행동)·Poisoning(검증되지 않은 사실) 재현/완화 시뮬레이션 추가.
3. (완료) **Tool Offloading/계층 설계 섹션**: “도구를 리트리벌로 로딩/제한”하거나 “상위 래퍼 도구로 단순화”하는 패턴 소개.
4. **‘삭제’ 중심의 운영 규칙**: 언제 넣고/빼고/격리할지 규칙(임계치, 주기, 스키마)과 로그 지표(토큰/비용/실패율) 추가.

3
context_engineering_research_agent/agent.py
View File

@@ -9,6 +9,7 @@ from typing import Any
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, FilesystemBackend, StateBackend
from langchain.agents.middleware.types import AgentMiddleware
from langchain.tools import ToolRuntime
from langchain_core.language_models import BaseChatModel
from langchain_openai import ChatOpenAI
@@ -212,7 +213,7 @@ def create_context_aware_agent(
routes={"/": local_fs_backend},
)
middlewares = []
middlewares: list[AgentMiddleware] = []
if enable_offloading:
offload_config = OffloadingConfig(

5
context_engineering_research_agent/backends/docker_session.py
View File

@@ -73,8 +73,11 @@ class DockerSandboxSession:
pids_limit=128,
working_dir=self.workspace_root,
)
container = self._container
if container is None:
raise RuntimeError("Docker 컨테이너 생성에 실패했습니다")
await asyncio.to_thread(
self._container.exec_run,
container.exec_run,
f"mkdir -p {self.workspace_root}/{META_DIR} {self.workspace_root}/{SHARED_DIR}",
)
except Exception as exc:

6
context_engineering_research_agent/context_strategies/isolation.py
View File

@@ -8,7 +8,7 @@ DeepAgents의 SubAgentMiddleware에서 task() 도구로 구현되어 있습니
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from typing import Any, NotRequired, TypedDict
from typing import Any, NotRequired, TypedDict, cast
from langchain.agents.middleware.types import (
AgentMiddleware,
@@ -82,10 +82,10 @@ class ContextIsolationStrategy(AgentMiddleware):
for spec in self._subagents:
if "runnable" in spec:
compiled = spec # type: ignore
compiled = cast(CompiledSubAgentSpec, spec)
agents[compiled["name"]] = compiled["runnable"]
elif self._agent_factory:
simple = spec # type: ignore
simple = cast(SubAgentSpec, spec)
agents[simple["name"]] = self._agent_factory(
model=simple.get("model", self.config.default_model),
system_prompt=simple["system_prompt"],

20
context_engineering_research_agent/context_strategies/reduction.py
View File

@@ -38,9 +38,11 @@ middleware = SummarizationMiddleware(
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import cast
from langchain.agents.middleware.types import (
AgentMiddleware,
AnyMessage,
ModelRequest,
ModelResponse,
)
@@ -278,13 +280,12 @@ class ContextReductionStrategy(AgentMiddleware):
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
"""모델 호출을 래핑하여 필요시 컨텍스트를 축소합니다."""
messages = list(request.state.get("messages", []))
messages = cast(list[BaseMessage], request.messages)
reduced_messages, result = self.reduce_context(messages)
if result.was_reduced:
modified_state = {**request.state, "messages": reduced_messages}
request = request.override(state=modified_state)
request = request.override(
messages=cast(list[AnyMessage], reduced_messages)
)
return handler(request)
@@ -294,13 +295,12 @@ class ContextReductionStrategy(AgentMiddleware):
handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
) -> ModelResponse:
"""비동기 모델 호출을 래핑합니다."""
messages = list(request.state.get("messages", []))
messages = cast(list[BaseMessage], request.messages)
reduced_messages, result = self.reduce_context(messages)
if result.was_reduced:
modified_state = {**request.state, "messages": reduced_messages}
request = request.override(state=modified_state)
request = request.override(
messages=cast(list[AnyMessage], reduced_messages)
)
return await handler(request)

3
pyproject.toml
View File

@@ -69,5 +69,6 @@ markers = [
[tool.mypy]
python_version = "3.13"
exclude = [".venv"]
exclude = ["(.venv|deepagents_sourcecode|research_workspace|tests|skills)"]
ignore_missing_imports = true
disable_error_code = ["import-untyped"]

35
research_agent/skills/middleware.py
View File

@@ -17,9 +17,11 @@
research_agent 프로젝트용으로 deepagents-cli에서 적응함.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import NotRequired, TypedDict, cast
from typing import Any, NotRequired, TypedDict, cast
from langchain.agents.middleware.types import (
AgentMiddleware,
@@ -27,6 +29,7 @@ from langchain.agents.middleware.types import (
ModelRequest,
ModelResponse,
)
from langchain_core.messages import SystemMessage
from langgraph.runtime import Runtime
from research_agent.skills.load import SkillMetadata, list_skills
@@ -177,8 +180,8 @@ class SkillsMiddleware(AgentMiddleware):
return "\n".join(lines)
def before_agent(
self, state: SkillsState, runtime: Runtime
) -> SkillsStateUpdate | None:
self, state: AgentState[Any], runtime: Runtime[Any]
) -> dict[str, Any] | None:
"""에이전트 실행 전에 스킬 메타데이터를 로드한다.
세션 시작 시 한 번 실행되어 사용자 레벨과 프로젝트 레벨
@@ -191,12 +194,13 @@ class SkillsMiddleware(AgentMiddleware):
Returns:
skills_metadata가 채워진 업데이트된 상태.
"""
_ = runtime
# 디렉토리 변경을 캐치하기 위해 각 상호작용마다 스킬 다시 로드
skills = list_skills(
user_skills_dir=self.skills_dir,
project_skills_dir=self.project_skills_dir,
)
return SkillsStateUpdate(skills_metadata=skills)
return {"skills_metadata": skills}
def wrap_model_call(
self,
@@ -215,7 +219,9 @@ class SkillsMiddleware(AgentMiddleware):
핸들러의 모델 응답.
"""
# 상태에서 스킬 메타데이터 가져오기
skills_metadata = request.state.get("skills_metadata", [])
skills_metadata = cast(
list[SkillMetadata], request.state.get("skills_metadata", [])
)
# 스킬 위치와 목록 포맷팅
skills_locations = self._format_skills_locations()
@@ -227,12 +233,13 @@ class SkillsMiddleware(AgentMiddleware):
skills_list=skills_list,
)
if request.system_prompt:
system_prompt = request.system_prompt + "\n\n" + skills_section
existing = str(request.system_message.content) if request.system_message else ""
if existing:
system_prompt = existing + "\n\n" + skills_section
else:
system_prompt = skills_section
return handler(request.override(system_prompt=system_prompt))
return handler(request.override(system_message=SystemMessage(content=system_prompt)))
async def awrap_model_call(
self,
@@ -250,7 +257,7 @@ class SkillsMiddleware(AgentMiddleware):
"""
# state_schema로 인해 상태가 SkillsState임이 보장됨
state = cast("SkillsState", request.state)
skills_metadata = state.get("skills_metadata", [])
skills_metadata = cast(list[SkillMetadata], state.get("skills_metadata", []))
# 스킬 위치와 목록 포맷팅
skills_locations = self._format_skills_locations()
@@ -262,10 +269,12 @@ class SkillsMiddleware(AgentMiddleware):
skills_list=skills_list,
)
# 시스템 프롬프트에 주입
if request.system_prompt:
system_prompt = request.system_prompt + "\n\n" + skills_section
existing = str(request.system_message.content) if request.system_message else ""
if existing:
system_prompt = existing + "\n\n" + skills_section
else:
system_prompt = skills_section
return await handler(request.override(system_prompt=system_prompt))
return await handler(
request.override(system_message=SystemMessage(content=system_prompt))
)

9
skills/academic-search/arxiv_search.py
View File

@@ -17,7 +17,6 @@ from __future__ import annotations
import argparse
import json
import sys
from typing import Any
@@ -42,8 +41,6 @@ def query_arxiv(
Output format: "text", "json", or "markdown" (default: "text").
Returns:
-------
str
The formatted search results or an error message.
"""
try:
@@ -97,9 +94,7 @@ def format_output(papers: list[dict[str, Any]], query: str, output_format: str)
output_format : str
Output format: "text", "json", or "markdown".
Returns
-------
str
Returns:
Formatted output string.
"""
if output_format == "json":
@@ -152,7 +147,7 @@ def format_output(papers: list[dict[str, Any]], query: str, output_format: str)
def main() -> None:
"""Main entry point for the arXiv search CLI."""
"""Run the arXiv search CLI."""
parser = argparse.ArgumentParser(
description="Search arXiv for academic research papers",
epilog="""

3
skills/skill-creator/scripts/discover_skills.py
View File

@@ -24,7 +24,6 @@ Exit Codes:
import argparse
import json
import re
import sys
from pathlib import Path
@@ -215,7 +214,7 @@ def get_default_skills_paths() -> list[Path]:
def main() -> int:
"""Main entry point."""
"""Run the main entry point."""
parser = argparse.ArgumentParser(
description="Discover and index existing skills for triage."
)

2
skills/skill-creator/scripts/validate_skill.py
View File

@@ -233,7 +233,7 @@ def validate_skill(skill_path: Path) -> ValidationResult:
def main() -> int:
"""Main entry point."""
"""Run the main entry point."""
if len(sys.argv) < 2:
print("Usage: python validate_skill.py <skill_path>")
print("Example: python validate_skill.py ~/.deepagents/skills/my-skill/")