자율적 Agent 코어 모듈의 5단계 설계와 「최소 실행 샘플」
최소 실행 샘플(agent_example.py)로 빠르게 이해해 보자.
빠르게 확인하고 싶은 경우에는 「7. 사용 예시(워크플로우)」부터 봐 주세요.
전체 소스 공개:
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"}.
개별 문서: 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)에 문의한다.
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