
「따라잡자!」 자율형 Agent를 「최소 실행 샘플 편」으로 빠르게 이해하기
요약
자율형 에이전트 GRACE의 코어 모듈 설계와 최소 실행 샘플을 통해 에이전트의 작동 원리를 설명합니다. 계획 생성부터 실행, 신뢰도 평가, 동적 리플랜까지 이어지는 5단계 설계 루프를 다룹니다.
핵심 포인트
- GRACE 에이전트의 5단계 설계 및 8개 코어 모듈 구조 설명
- Planner, Executor, Memory를 포함한 자율 루프 메커니즘
- 신뢰도 기반의 인간 개입(HITL) 및 동적 리플랜 전략
- Anthropic Claude 및 Gemini Embedding 활용 기술 스택

자율적 Agent 코어 모듈의 5단계 설계와 「최소 실행 샘플」
최소 실행 샘플(agent_example.py)로 빠르게 이해해 보자.
빠르게 확인하고 싶은 경우에는 「7. 사용 예시(워크플로우)」부터 봐 주세요.
전체 소스 공개:
- Claude API 버전 → https://github.com/nakashima2toshio/anthropic_grace_agent_v2
- 로컬 LLM 버전 → https://github.com/nakashima2toshio/ollama_grace_agent_v2
Version 1.1| 최종 업데이트: 2026-06-28
목차
- 개요
-
- 아키텍처 구성도
-
- 모듈 구성도
-
- 모듈별 요약 (클래스·함수 목록)
-
- 실행 메모리가 쌓이기까지 (planner → executor → memory)
-
- 처리 시퀀스 (GRACE 루프)
-
- 설정·상수 (횡단)
-
- 사용 예시 (워크플로우)
-
- 엑스포트
-
- 변경 이력
- 부록: 의존 관계도
「최소 실행 샘플은: agent_example.py」
- A. 코어의 기본 구성: 자율형 Agent의 5단계 설계
- B. 구현의 코어 모듈 구성 (8개 코어 모듈)
- C. 역할 요약
- D. 최소 실행 샘플 agent_example.py
- E. 프롬프트와 API 발행부
- F. 이해를 위한 보충 설명
개요
본 문서는 GRACE (Guided Reasoning with Adaptive Confidence Execution) 자율 에이전트의 **코어 모듈군 (A 그룹)**을 횡단적으로 조망하기 위한 정리 문서이다. 8개의 프로그램이 「계획 생성 → 스텝 실행 → 신뢰도 평가 → 교정 → 개입 (HITL) → 동적 리플랜(Replan)」이라는 루프를 구성하며, executor.py를 오케스트레이터(Orchestrator)로 하여 연계한다.
각 모듈의 IPO 상세 (시그니처·반환값 예시·사용 예시)는 개별 문서에 맡기고, 본서는 전체 아키텍처·데이터 플로우·모듈 간 연계·링크 모음에 집중한다.
📝
기술 스택: LLM 용도는 모두 Anthropic Claude (기본 claude-sonnet-4-6, 경량 claude-haiku-4-5-20251001, 키 ANTHROPIC_API_KEY). 검색의 Embedding만 Gemini gemini-embedding-001 (3072 차원, 키 GOOGLE_API_KEY)를 계속 이용한다. LLM 클라이언트는 grace.llm_compat.create_chat_client()를 경유한다.
주요 책무
- 사용자 쿼리로부터 실행 계획 (
ExecutionPlan)을 생성한다 (복잡도 추정·모호한 쿼리 탐지·이층 방식) - 계획을 순차적으로 실행하며, 도구 호출(Tool Call)·의존성 해결·동적 폴백(Fallback)을 관리한다
- 스텝별·전체의 다축 신뢰도를 계산한다 (검색 품질·LLM 자기 평가·소스 일치·근거 타당성)
- 가공되지 않은 신뢰도를 경험적 정답률로 온도 스케일링(Temperature Scaling)을 통해 교정한다
- 과거 실행 실적을 축적하여 컬렉션 우선순위를 학습한다
- 신뢰도에 기반하여 인간 개입 (HITL)을 단계적으로 게이트(Gate)한다
- 실패·저신뢰도 시 전략을 선택하여 계획을 동적으로 수정한다 (리플랜)
- RAG 검색·Web 검색·LLM 추론·사용자 질문 도구를 통일된 인터페이스로 제공한다
각 책무 대응 모듈
| # | 책무 | 대응 모듈 | 설명 |
|---|---|---|---|
| 1 | 실행 계획 생성 | planner.py | 복잡도 추정·이층 방식(Rule/LLM)·모호한 쿼리 탐지로 ExecutionPlan을 생성 |
| 2 | 계획의 순차 실행·통괄 | executor.py | 스텝 실행 루프·동적 폴백(Fallback)·전체 컴포넌트 연계의 중추 |
| 3 | 다축 신뢰도 계산 | confidence.py | 검색 품질/LLM 자기 평가/소스 일치/근거 타당성을 통합하여 개입 레벨을 결정 |
| 4 | 신뢰도 교정 | calibration.py | 온도 스케일링(Temperature Scaling)을 통해 생점수(Raw Score)를 경험적 정답률에 가깝게 조정 |
| 5 | 실행 메모리 축적·학습 | memory.py | JSONL에 실적을 축적하여 쿼리별 컬렉션 사전 분포를 제공 |
| 6 | 인간 개입 (HITL) | intervention.py | SILENT/NOTIFY/CONFIRM/ESCALATE 게이트 및 동적 임계값 학습 |
| 7 | 동적 리플랜 (Replan) | replan.py | 실패·저신뢰도·피드백을 계기로 전략 선택을 통한 재계획 |
| 8 | 도구 실행 | tools.py | RAG 검색·Web 검색·추론·ask_user를 ToolRegistry로 통합 제공 |
주요 기능 목록
| 기능 | 설명 |
|---|---|
Planner / create_planner() | 실행 계획 생성 에이전트 및 해당 팩토리 |
Executor / ExecutionState / create_executor() | 실행 오케스트레이터·실행 상태·팩토리 |
ConfidenceCalculator / ConfidenceAggregator | 스텝 신뢰도 산출·전체 집계 |
LLMSelfEvaluator / SourceAgreementCalculator / QueryCoverageCalculator / GroundednessVerifier | 신뢰도의 각 축 평가기 |
Calibrator / fit_temperature() | 온도 스케일링을 통한 신뢰도 교정 |
ExecutionMemory / create_execution_memory() | 실행 메모리 계층 (실적 축적·우선 컬렉션 학습) |
InterventionHandler / DynamicThresholdAdjuster / ConfirmationFlow | 개입 처리·동적 임계값 조정·확인 플로우 |
ReplanManager / ReplanOrchestrator | 리플랜 판정·전략 결정·Executor 통합 |
ToolRegistry / RAGSearchTool / WebSearchTool / ReasoningTool / AskUserTool | 도구 레지스트리 및 각 도구 |
1. 아키텍처 구성도
1.0 모듈·블록도 (전체 처리 플로우)
A 그룹의 8개 프로그램을 각각 1개 블록으로 나타내어, 사용자 쿼리부터 최종 답변까지의 전체 처리 플로우를 보여준다. executor.py가 중추가 되어 각 모듈과 왕복하며 루프를 구동한다.
1.1 시스템 전체 구성 (3계층)
1.2 데이터 플로우
- 클라이언트 계층 (UI / 벤치마크 / API)이
query를 입력한다. planner.py가 복잡도를 추정하며, 모호한 쿼리라면 확인 계획(ask_user)을, 그 외에는 규칙 기반(Rule-based) 또는 LLM 계획으로서ExecutionPlan을 생성한다 (memory.py의 사전 분포를 반영하여 우선 컬렉션 반영).executor.py가 계획을 받아 의존성을 해결하며 스텝을 순차적으로 실행한다. 각 스텝은tools.py의ToolRegistry.execute()를 통해 도구를 호출한다.- RAG 검색 점수와 적합성에 따라,
executor.py가web_search→ask_user를 동적으로 삽입/스킵한다. - 각 스텝 이후,
confidence.py가ConfidenceFactors로부터ConfidenceScore를 [계산한다]
를 산출하고, ActionDecision (개입 레벨)을 결정한다. - 전체 신뢰도를 ConfidenceAggregator에서 집계하고, calibration.py의 Calibrator.transform()으로 온도 교정 (Temperature Calibration)한다. - intervention.py가 개입 레벨에 따라 자동 진행/알림/확인/에스컬레이션 (Escalation)을 게이트 (Gate)한다. 피드백은 DynamicThresholdAdjuster가 임계값 학습에 반영한다. - 스텝 실패나 저신뢰도 시에는 replan.py가 전략 (FULL/PARTIAL/FALLBACK/SKIP/ABORT)을 선택하여 planner.py로 위임해 새로운 계획을 생성하고, executor.py가 재귀 실행한다. - executor.py가 실행 실적 (사용 컬렉션·성패·신뢰도)을 memory.py에 기록하고, 최종적으로 ExecutionResult를 반환한다.
2. 모듈 구성도
A 그룹 내부의 주요 클래스와 연계를 나타낸다.
2.1 모듈 간 의존 관계 테이블
| 모듈 | 주로 호출하는 상대 | 주로 호출되는 상대 |
|---|---|---|
planner.py | memory (사전 분포), llm_compat, schemas, services.qdrant_service | executor, replan, UI |
executor.py | tools, confidence, calibration, intervention, replan, memory | UI, benchmark |
tools.py | Qdrant, Gemini Embedding, Web 검색, llm_compat | executor |
confidence.py | llm_compat (Anthropic), Gemini Embedding | executor |
calibration.py | (stdlib 만) | executor, 평가 스크립트 |
memory.py | (stdlib 만·JSONL) | executor (쓰기), planner (읽기) |
intervention.py | confidence (InterventionLevel / ActionDecision) | executor |
replan.py | planner (create_plan / 위임) | executor |
3. 모듈별 요약 (클래스·함수 목록)
각 모듈의 책임·주요 클래스·함수를 요약한다. IPO 상세 내용은 각 「개별 문서」를 참조한다.
3.1 planner.py — 계획 생성
개별 문서: planner.md
사용자의 질문으로부터 ExecutionPlan을 생성한다. 이층 방식 (복잡도 < 0.7일 때 규칙 기반 (Rule-based), ≥ 0.7 또는 Web 검색 마커가 있을 때 LLM 계획)과 모호한 쿼리 탐지를 갖추고 있다.
| 요소 | 개요 |
|---|---|
Planner | 계획 생성 에이전트 본체 |
Planner.create_plan(query) | 질문으로부터 ExecutionPlan을 생성 (이층 방식) |
Planner.estimate_complexity(query) | 키워드 기반으로 복잡도를 0.0–1.0 추정 |
Planner.estimate_complexity_with_llm(query) | LLM으로 복잡도를 추정 |
Planner.refine_plan(plan, feedback) | 피드백에 기반하여 계획을 수정 |
is_ambiguous_query(query) | 지시어만 있고 대상이 불분명한 모호한 쿼리를 판정 |
create_planner(config, model_name) | Planner 팩토리 |
주요 상수: PLAN_GENERATION_PROMPT, COMPLEXITY_ESTIMATION_PROMPT, _COMPLEXITY_FACTORS, _AMBIGUOUS_REFERENT_PATTERNS
, _LLM_PLAN_MARKERS
임계값 llm_plan_complexity_threshold=0.7
3.2 executor.py — 실행 오케스트레이터 (Execution Orchestrator)
개별 문서: executor.md
생성된 ExecutionPlan을 순차적으로 실행하는 중추. 도구 실행(Tool execution), 동적 폴백(Dynamic fallback), 신뢰도 계산, 교정(Calibration), HITL(Human-in-the-loop) 개입, 리플랜(Replan) 연계를 총괄하며 최종 답변을 생성한다.
| 요소 | 개요 |
|---|---|
Executor | 계획 실행 에이전트 (GRACE 네이티브 구현) |
Executor.execute_plan(plan) | 블로킹(Blocking) 실행으로 ExecutionResult를 반환 |
Executor.execute_plan_generator(plan, state) | UI 연동용 제너레이터 버전 (중간 이벤트를 yield) |
Executor.execute(plan) | 통합 엔트리 포인트 (benchmark 호환) |
Executor.cancel(state) / resume(state) | 실행 제어 |
ExecutionState | 실행 상태 (계획, 스텝 결과, 신뢰도, 제어 플래그) |
create_executor(config, tool_registry, ...) | Executor 팩토리 |
주요 설정: parallel_search, max_parallel_steps=3, react_enabled, rag_sufficient_score=0.7, max_replans=3, calibration_path. _SEARCH_ACTIONS = ("rag_search", "web_search").
3.3 confidence.py — 신뢰도 계산 (Confidence Calculation)
개별 문서: confidence.md
검색 품질, LLM 자기 평가, 소스 일치도, 근거 타당성 (Groundedness)의 다축을 통합하여 신뢰도 점수와 개입 레벨 (SILENT/NOTIFY/CONFIRM/ESCALATE)을 결정한다.
| 요소 | 개요 |
|---|---|
ConfidenceCalculator | 하이브리드 방식으로 점수 산출 및 decide_action()으로 개입 레벨 결정 |
LLMSelfEvaluator | LLM을 통한 확신도, 망라성, Factors 통합 평가 |
SourceAgreementCalculator | Gemini Embedding으로 소스 간 코사인 유사도(Cosine similarity) 산출 |
QueryCoverageCalculator | 쿼리 망라도를 LLM으로 0.0–1.0 평가 |
GroundednessVerifier | 각 주장의 인용 지지/모순을 판정하여 지지율을 반환 (S1 핵심) |
ConfidenceAggregator | 여러 스텝을 mean/min/weighted 방식으로 집계 |
ConfidenceFactors / ConfidenceScore / ActionDecision / InterventionLevel | 입력 요소, 점수, 개입 결정, 레벨 열거 |
주요 임계값: silent=0.9, notify=0.7, confirm=0.4. 가중치 합계 1.0 (search_quality=0.25 외).
3.4 calibration.py — 신뢰도 교정 (Confidence Calibration)
개별 문서: calibration.md
가공되지 않은 신뢰도와 경험적 정답률 사이의 괴리 (ECE, Expected Calibration Error)를 온도 스케일링 (logit → /T → sigmoid)을 통해 축소한다. SciPy에 의존하지 않는 1차원 탐색으로 온도 T를 추정하고 JSON에 영속화한다.
| 요소 | 개요 |
|---|---|
Calibrator | 온도 스케일링 교정기 (dataclass) |
Calibrator.transform(p) | 단일 신뢰도에 교정 적용 |
Calibrator.fit(confidences, correctness) | (신뢰도, 정오)로부터 T를 추정하여 생성 |
Calibrator.save(path) / load(path) | JSON 영속화 및 로드 |
apply_temperature(p, temperature) | 온도 T 적용 |
fit_temperature(confidences, correctness) | 2단계 그리드 탐색으로 이진 NLL 최소화 |
expected_calibration_error(...) | 등간격 빈(bin)으로 ECE 산출 |
주요 상수: DEFAULT_CALIBRATION_PATH="config/calibration.json", _EPS=1e-6.
퇴화 데이터는 항등 교정 (T=1.0)으로 폴백(fallback).
3.5 memory.py — 실행 메모리
개별 문서: (신규 · 본서에서 최초 등장)
실행 레코드를 JSONL에 축적하고, 컬렉션별 성공률·평균 신뢰도 (Laplace 평활화)를 집계하여, 쿼리 키워드에 기반한 컬렉션 사전 분포 (collection prior distribution) 를 planner.py에 제공한다 (에피소드 기억).
| 요소 | 개요 |
|---|---|
ExecutionMemory | JSONL 스토리지 및 우선순위 계산의 주 API |
ExecutionMemory.record(query, collection, success, confidence) | 1건의 실적을 추가 |
ExecutionMemory.collection_priors(query) | 쿼리별 우선 컬렉션 통계를 반환 |
ExecutionMemory.best_collection(query) | 충분한 실적을 가진 최적의 컬렉션을 추천 |
MemoryRecord / CollectionStat | 레코드 / 집계 통계의 dataclass |
extract_keywords(text) | 형태소 분석 없는 경량 키워드 추출 |
create_execution_memory(path) | ExecutionMemory 팩토리 |
주요 상수: DEFAULT_MEMORY_PATH="logs/grace_memory.jsonl".
표준 라이브러리(stdlib)만 사용하며 네트워크에 의존하지 않음.
3.6 intervention.py — 개입 (HITL)
개별 문서: intervention.md
confidence.py의 ActionDecision을 받아, 4단계 (SILENT/NOTIFY/CONFIRM/ESCALATE)로 실행을 게이트(gate)하는 Human-in-the-Loop 계층. 사용자 피드백을 통해 임계값(threshold)을 동적으로 학습한다.
| 요소 | 개요 |
|---|---|
InterventionHandler | 개입 레벨에 따라 요청을 생성 및 디스패치 |
InterventionHandler.handle(decision, step, plan) | 개입을 처리하고 InterventionResponse를 반환 |
DynamicThresholdAdjuster | FP/FN 비율을 모니터링하여 임계값을 동적 조정 |
DynamicThresholdAdjuster.record_feedback(confidence, was_correct) | 피드백을 기록하고 임계값 학습 |
ConfirmationFlow | 확인 → 수정 → 재확인 루프를 최대 횟수까지 관리 |
InterventionRequest / InterventionResponse / InterventionAction / FeedbackRecord | 요청 · 응답 · 액션 열거형(enum) · 피드백 기록 |
create_intervention_handler() / create_threshold_adjuster() / create_confirmation_flow() | 각 팩토리 |
주요 상수: timeout_seconds=300, learning_rate=0.05, min_samples=10. FP/FN 비율 > 30%일 때 임계값(threshold) 조정.
3.7 replan.py — 동적 리플랜 (Dynamic Replan)
개별 문서: replan.md
단계 실패, 낮은 신뢰도, 사용자 피드백을 계기로 전략(FULL/PARTIAL/FALLBACK/SKIP/ABORT)을 선택하여 계획을 동적으로 수정한다. 새로운 계획 생성은 planner.py로 위임한다.
| 요소 | 개요 |
|---|---|
ReplanOrchestrator | Executor와 ReplanManager를 통합하여 자동 리플랜 플로우를 관리 |
ReplanOrchestrator.handle_step_failure(...) | 단계 실패 시의 리플랜 처리 |
ReplanOrchestrator.handle_user_feedback(...) | 피드백에 의한 리플랜 처리 |
ReplanManager | 리플랜 필요 여부 판정·전략 결정·새 계획 생성 |
ReplanManager.should_replan(step_result, replan_count) | 단계 결과로부터 필요 여부 및 트리거 판정 |
ReplanManager.determine_strategy(context, plan) | 전략을 결정 |
ReplanTrigger / ReplanStrategy / ReplanContext / ReplanResult | 트리거/전략의 열거형(Enum) 및 컨텍스트/결과 |
create_replan_manager() / create_replan_orchestrator() | 각 팩토리 |
주요 상수: max_replans=3, confidence_threshold=0.4, partial_replan_threshold=0.6, _SEARCH_FALLBACK_CHAIN={"rag_search":"web_search","web_search":"rag_search"}.
3.8 tools.py — 도구군 (Tools)
개별 문서: tools.md
GRACE 에이전트의 통합 도구 시스템. RAG 검색 (Gemini Embedding + Qdrant)·Web 검색·LLM 추론 (Anthropic Claude)·ask_user (HITL)를 ToolRegistry에서 이름 기반으로 실행한다.
| 요소 | 개요 | ActionType |
|---|---|---|
ToolRegistry | 도구 등록·취득·실행 디스패치 | — |
ToolRegistry.execute(name, **kwargs) | 이름 지정으로 도구를 실행 | — |
BaseTool | 모든 도구의 추상 기저 클래스 (execute()) | — |
RAGSearchTool | Qdrant로부터 RAG 검색 (동적 컬렉션·임계값 조정) | rag_search |
WebSearchTool | SerpAPI/DDG/Google CSE를 이용한 Web 검색 | web_search |
ReasoningTool | Anthropic Claude를 이용한 답변 생성 | reasoning |
AskUserTool | HITL를 통해 질문을 구조화하여 반환 | ask_user |
ToolResult | 성공·출력·신뢰도·에러·실행 시간의 통합 결과 | — |
create_tool_registry(config) | ToolRegistry 팩토리 | — |
주요 상수: 추론 LLM claude-sonnet-4-6, Embedding gemini-embedding-001 (3072 차원), Qdrant http://localhost:6333, 유효 도구 ["rag_search","web_search","reasoning","ask_user"].
4. 실행 메모리가 쌓일 때까지 (planner → executor → memory)
본 장에서는 planner.py에서 시작하여 executor.py
의 실행 결과가 memory.py의 ExecutionMemory에 저장되는 과정을 예시 데이터와 경우의 수를 통해 친절하게 설명한다. 핵심 요점은 **"메모리는 executor가 쓰고, planner가 읽는다"**라는 일방통행식 학습 루프이며, 이번 실행 결과가 다음 회차 이후의 계획 수립에 영향을 미친다는 점에 있다.
4.1 등장인물 (전체 모습과 시퀀스 다이어그램)
처리 골격은 다음과 같다.
사용자의 질문
│
▼
...
4.2 예시 데이터 설정
- Qdrant의 컬렉션 (Collection) 후보:
wikipedia_ja,livedoor,cc_news,japanese_text - 사용자는 "Python의 ◯◯"라는 질문을 몇 번 반복한다고 가정한다.
- 메모리 파일
logs/grace_memory.jsonl은 처음에는 비어 있다고 가정한다.
4.3 planner.py: 계획 수립 전 메모리에 문의
planner.py:232의 _prioritized_collection()이 메모리 (memory)에 문의한다.
# grace/planner.py:232
def _prioritized_collection(self, query: str) -> Optional[str]:
if self._memory is None:
...
🔀 경우의 수 ①: planner가 컬렉션을 좁힐 수 있는가
| 상황 | _prioritized_collection()의 반환값 | 계획의 rag_search |
|---|---|---|
메모리 비활성화 (config.memory.enabled=false) | None | 모든 컬렉션 검색 |
| 실적은 있으나 불충분 (건수 < 3 또는 스코어 < 0.6) | None | 모든 컬렉션 검색 |
| 실적이 충분 (건수 ≥ 3 이고 스코어 ≥ 0.6) | 예: "wikipedia_ja" | 해당 컬렉션으로 한정하여 검색 |
📝 처음에는 메모리가 비어 있으므로 반드시 None (= 모든 컬렉션 검색)이 된다. "처음 몇 번은 탐색하며, 데이터가 쌓이면 똑똑해진다"라는 설계이다.
4.4 executor.py: 사용 컬렉션의 수집
계획을 실행하는 도중, rag_search
AI 자동 생성 콘텐츠
본 콘텐츠는 Zenn AI의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.
원문 바로가기