TL;DR

배경: LLM의 물리적 한계와 Position Bias

컨텍스트 윈도우: LLM이 한 번에 읽을 수 있는 최대 토큰 길이 (예: Claude는 200K 토큰)
Position Bias: 초반부 텍스트에 과도한 가중치를 주는 모델의 편향
Lost-in-the-Middle: 긴 문서의 중앙부 정보가 덜 활용되는 현상

대규모 언어 모델의 컨텍스트 윈도우는 지속적으로 확장되고 있으나, 실제 사용 성능은 선형적으로 따라가지 못한다. Liu et al. (2023)의 연구에 따르면, 다중 문서 질의응답 환경에서 컨텍스트 중앙부의 정보 활용도가 20-30% 저하되는 Lost-in-the-Middle 현상이 발생한다.

더욱 심각한 문제는 Position Bias다. 모델은 초기 토큰에 과도한 가중치를 할당하는 구조적 편향이 있으며, 이는 논문 요약이나 법률 문서 분석에서 초반부 정보에 편향된 최대 35%의 정보 왜곡을 유발한다. 기존의 선형적 청크-요약-병합 파이프라인은 이 편향을 순차적으로 전파할 뿐, 능동적으로 상쇄하지 못한다.

Loop Engineering은 이 한계를 인식하고, 단일 패스의 선형적 처리가 아닌 다각도의 재진입을 통해 편향을 통계적으로 중화하는 구조적 전환을 제안한다.

핵심 메커니즘: 상태 유지형 다중 패스와 재귀적 정제

Loop Engineering의 본질은 LLM 호출 간 외부 상태 버퍼(External State Buffer) 를 유지하며, 각 패스의 중간 산출물을 다음 패스의 동적 누적 컨텍스트(Dynamic Accumulated Context)로 주입하는 데 있다. 이는 정적인 검색-주입(Retrieval-Augmented Generation) 방식과 근본적으로 다르다.

# Loop Engineering 핵심 의사코드
상태 S_0 = {}
for i in 1..N:
    # 순열 기반 청킹 등 전략에 따른 청크 집합 선정
    청크_집합 C_i = select_chunks(document, strategy=i)
    
    # 이전 상태를 컨텍스트로 주입하여 LLM 호출
    결과 R_i = LLM(prompt=C_i, context=S_{i-1})  
    
    # 충돌 플래깅 후 상태 병합 (단순 덮어쓰기 지양)
    상태 S_i = merge_with_conflict_detection(S_{i-1}, R_i)  

위 수식에서 핵심은 merge 연산의 방향성이다. 이전 패스의 결론과 현재 패스의 결과가 상충할 경우, 우선순위 기반 덮어쓰기가 아닌 충돌 플래깅 후 재처리(Flag-and-Reprocess) 전략을 취한다. 충돌이 감지된 엔티티 그래프의 서브그래프는 다음 패스에서 양측의 증거가 동시에 제시된 상태로 재평가된다.

이러한 다층 루프 구조는 탐색, 정제, 교차 검증의 3단계로 세분화된다. 1차 탐색 패스에서는 의존성 그래프를 구축하고, 내부 정제 패스에서는 위상 정렬된 순서로 하위 모듈에 상위 모듈의 컨텍스트를 주입한다. 특히 주목할 점은 부분 재처리(Partial Re-processing) 최적화다. 전체 문서를 재평가하지 않고 불일치가 발견된 서브그래프만을 추가 패스에 투입함으로써, 토큰 비용의 증가를 약 2배 수준으로 억제한다 (로그 스케일이 아닌 상수배 최적화).

비용 프로필 예시:

Position Bias 상쇄를 위해 도입된 순열 기반 다중 청킹(Permutation-based Multi-chunking) 은 동일 문서를 K개의 서로 다른 순서로 재배열하여 독립 요약을 생성한다. 이후 합의 기반 병합(Consensus-based Merging)을 통해 편향을 8-12% 수준까지 감소시킨다. 또한, 슬라이딩 윈도우 중첩 청킹(30-50% 중첩)을 통해 경계 효과를 완화하며, 최종적으로 역방향 수반 관계 검증(Reverse Entailment Check) 을 통해 요약 문장 S가 원본 청크 C에서 논리적으로 도출되지 않으면 환각으로 간주하고 재처리 플래그를 설정한다.

한계 및 제약: 오류 증폭과 수렴의 불확실성

Loop Engineering은 컨텍스트의 논리적 확장을 가능하게 하지만, 치명적인 부작용을 수반한다. 가장 위험한 것은 오류 증폭 루프(Error Amplification Loop) 이다. 초기 패스에서 발생한 환각이 외부 상태 버퍼에 기록되면, 후속 패스는 이를 ‘기존 지식’으로 전제하여 새로운 오류를 생산하거나 기존 오류를 강화한다. 단일 패스 환각과 달리 구조화된 상태에 기반한 오류는 탐지와 롤백이 극히 어렵다.

또한, 반복 패스의 수렴에 대한 이론적 보장이 부재하다. 경험적으로 3-5회 패스에서 수렴하며 6회 이상에서는 한계 효용(Marginal Gain)이 3% 미만으로 급락하지만, 법률이나 의학과 같은 고전문성 도메인에서는 패스 간 결론이 진동(Oscillation)하거나 발산할 가능성이 존재한다. 이는 상태 병합 로직의 휴리스틱 의존도가 높음을 시사한다.

비용과 지연 시간의 문제도 명확하다. 단일 패스 대비 3-5배의 API 호출이 발생하므로 실시간성이 요구되는 서비스에는 부적합하다. 더욱이 장시간 루프 실행 시 LLM 에이전트의 자체 컨텍스트 관리로 인해 상태가 손실될 수 있으므로, 위에서 다룬 명시적 체크포인트 저장 전략이 필수적이다.

실무 적용: 한국 백엔드/ML 개발자를 위한 가이드

국내 대형 IT 기업의 RAG 파이프라인 운영 팀은 이미 유사 패턴을 도입 중이다. 문서 임베딩 기반 정적 검색으로는 해결되지 않는 대규모 계약서 분석, 특허 문헌 조사, 기술 문서 마이그레이션 등의 태스크에서 Loop Engineering 원칙을 적용해 청킹 전략과 검증 루프를 재설계한 사례가 보고되고 있다.

1. Loop Engineering의 Python 구현

Claude Code는 Python 에이전트이므로, Loop Engineering은 명시적 상태 관리로 구현됩니다.

#### 핵심 구조

import json
from anthropic import Anthropic
from datetime import datetime
class LoopEngineer:
    def __init__(self, model: str = "claude-opus-4-1", max_passes: int = 5):
        self.client = Anthropic()
        self.model = model
        self.max_passes = max_passes
        self.state_history = []
        
    def run(self, documents: list[str], task_description: str) -> dict:
        """Loop Engineering 메인 루프"""
        state = {}
        
        for pass_num in range(1, self.max_passes + 1):
            print(f"\n{'='*60}")
            print(f"Pass {pass_num}/{self.max_passes} | {datetime.now().strftime('%H:%M:%S')}")
            print(f"{'='*60}")
            
            # Step 1: 청크 선택 (다양한 순열로 Position Bias 상쇄)
            chunks = self._select_chunks(documents, pass_num)
            print(f"Selected {len(chunks)} chunks | Strategy: {self._strategy_name(pass_num)}")
            
            # Step 2: LLM 호출 (이전 상태 주입)
            results = self._llm_call(chunks, state, task_description)
            print(f"LLM processing completed | Output tokens: {len(results.get('analysis', '').split())}")
            
            # Step 3: 충돌 탐지 및 상태 병합
            conflicts = self._detect_conflicts(state, results)
            state = self._merge_with_conflict_handling(state, results, conflicts)
            print(f"State merged | Conflicts: {len(conflicts)} | State size: {len(json.dumps(state))}")
            
            # Step 4: 체크포인트 저장 (중간 손실 방지)
            checkpoint_path = f"loop_checkpoint_pass_{pass_num}.json"
            self._save_checkpoint(state, checkpoint_path)
            self.state_history.append({
                "pass": pass_num,
                "state_size": len(json.dumps(state)),
                "conflicts": len(conflicts),
                "checkpoint": checkpoint_path
            })
            
            # Step 5: 수렴 확인
            if self._check_convergence(pass_num):
                print(f"\n✓ Converged at Pass {pass_num}. Terminating early.")
                break
        
        print(f"\n{'='*60}")
        print(f"Loop Engineering Complete | Total Passes: {pass_num}")
        print(f"Final State Size: {len(json.dumps(state))} bytes")
        print(f"{'='*60}\n")
        
        return state
    
    def _select_chunks(self, documents: list[str], pass_num: int) -> list[str]:
        """순열 기반 청킹 (Position Bias 상쇄)"""
        chunks = []
        chunk_size = 1024
        overlap = int(chunk_size * 0.4)  # 40% 중첩
        
        for doc in documents:
            for i in range(0, len(doc), chunk_size - overlap):
                chunk = doc[i:i+chunk_size]
                if chunk:
                    chunks.append(chunk)
        
        # 패스별 재배열 전략
        if pass_num % 4 == 1:
            return chunks  # Forward
        elif pass_num % 4 == 2:
            return chunks[::-1]  # Reverse
        elif pass_num % 4 == 3:
            mid = len(chunks) // 2
            return chunks[mid:] + chunks[:mid]  # Center-first
        else:  # pass_num % 4 == 0
            even = chunks[::2]
            odd = chunks[1::2]
            return [c for pair in zip(even, odd) for c in pair]  # Alternate
    
    def _strategy_name(self, pass_num: int) -> str:
        names = ["Forward", "Reverse", "Center-First", "Alternate"]
        return names[(pass_num - 1) % 4]
    
    def _llm_call(self, chunks: list[str], context: dict, task: str) -> dict:
        """Claude API 호출"""
        context_str = json.dumps(context, ensure_ascii=False, indent=2) if context else "{}"
        
        prompt = f"""Task: {task}
Previous Analysis (from prior passes):
{context_str}
New Document Batch:
{chr(10).join([f"--- Chunk {i+1} ---{chr(10)}{chunk[:200]}..." for i, chunk in enumerate(chunks[:5])])}
(and {len(chunks)-5} more chunks)
Instructions:
1. Analyze each new chunk in context of previous analysis
2. Identify any conflicts with previous conclusions
3. Flag conflicts for re-evaluation in next pass
4. Output structured JSON with: {{"analysis": "...", "entities": {{...}}, "conflicts": [...]}}
Output JSON:"""
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return {"analysis": response.content[0].text, "entities": {}, "conflicts": []}
    
    def _detect_conflicts(self, old_state: dict, new_results: dict) -> list[dict]:
        """이전 결과와 현재 결과 간의 충돌 탐지"""
        conflicts = []
        
        old_entities = old_state.get("entities", {})
        new_entities = new_results.get("entities", {})
        
        for key in set(old_entities.keys()) & set(new_entities.keys()):
            if old_entities[key] != new_entities[key]:
                conflicts.append({
                    "entity": key,
                    "previous": old_entities[key],
                    "current": new_entities[key],
                    "flag_for_reprocess": True
                })
        
        return conflicts
    
    def _merge_with_conflict_handling(self, state: dict, results: dict, conflicts: list[dict]) -> dict:
        """충돌 플래깅 후 상태 병합"""
        merged = state.copy()
        
        # 충돌이 없는 엔티티는 바로 업데이트
        for key, value in results.get("entities", {}).items():
            if not any(c["entity"] == key for c in conflicts):
                merged.setdefault("entities", {})[key] = value
        
        # 충돌이 있는 엔티티는 플래깅하여 다음 패스에서 재처리
        merged["reprocess_flags"] = conflicts
        
        return merged
    
    def _save_checkpoint(self, state: dict, path: str):
        """상태를 파일에 저장 (중간 손실 방지)"""
        with open(path, "w") as f:
            json.dump(state, f, indent=2, ensure_ascii=False)
    
    def _check_convergence(self, pass_num: int) -> bool:
        """수렴 조건: Pass 3 이후 상태 변화 < 3% or 충돌 0건"""
        if pass_num < 3:
            return False
        
        if pass_num >= 2:
            curr_size = self.state_history[-1].get("state_size", float("inf"))
            prev_size = self.state_history[-2].get("state_size", 0) if len(self.state_history) > 1 else 0
            
            size_change = abs(curr_size - prev_size) / (prev_size + 1)
            conflicts = self.state_history[-1].get("conflicts", 0)
            
            # 수렴 조건: (a) 상태 변화 < 3% AND (b) 충돌 < 2건
            return size_change < 0.03 and conflicts < 2
        
        return False

사용 예시
if __name__ == "__main__":
    engineer = LoopEngineer(max_passes=5)
    
    documents = [
        "Sample paper 1 abstract and content...",
        "Sample paper 2 abstract and content...",
    ]
    
    final_state = engineer.run(
        documents=documents,
        task_description="Analyze papers and extract: authors, publication year, main contributions"
    )
    
    print("Final Analysis:")
    print(json.dumps(final_state, ensure_ascii=False, indent=2))

#### 컨텍스트 손실 방지 전략

Claude의 자동 컨텍스트 압축은 기본 동작이며 사용자가 직접 조정할 수 없습니다. 대신 다음 전략을 권합니다:

1. 명시적 체크포인트 (위 코드 _save_checkpoint)
– 각 패스마다 상태를 JSON으로 저장
– 루프 중단 시에도 마지막 체크포인트부터 복구 가능

2. Prompt Caching 활용 (고급)

   # 반복되는 컨텍스트 부분을 캐싱하여 비용 감소
   system = [
       {
           "type": "text",
           "text": "You are a document analysis expert...",
           "cache_control": {"type": "ephemeral"}
       }
   ]
   ``

3. 토큰 카운팅 (모니터링)
   `python
   # 각 패스의 토큰 사용량 추적
   input_tokens = response.usage.input_tokens
   output_tokens = response.usage.output_tokens
   print(f"Pass {pass_num}: {input_tokens + output_tokens} tokens")
   `

2. 루프 설계 원칙: 3가지 체크리스트
Loop Engineering의 효과는 루프 설계에 달려 있습니다.
#### 원칙 1: 명확한 역할 분담 (각 패스의 목표 명시)

#### 원칙 2: 충돌 해결 지침

#### 원칙 3: 부분 재처리 트리거

3. 실전 예시: 논문 100개 분석

예상 비용 및 시간 (100개 논문 기준):

4. 선택지: 언제 Loop Engineering을 쓸 것인가?
아래 시나리오별 추천:

5. 비용 최적화 팁
Tip 1: 모델 선택
Claude 3.5 Sonnet ($3/MTok input, $15/MTok output)
Claude 3.5 Haiku ($0.80/MTok input, $4/MTok output)
Haiku로 루프 실행 시: 약 1/4 비용 (정확성 다소 저하)
Tip 2: Prompt Caching

Tip 3: 부분 재처리 임계값 조정
충돌 상한: 2-3건 (엄격) vs 5건 (관대)
신뢰도 임계값: 60% (엄격) vs 40% (관대)
더 관대할수록 Pass 수 증가 → 비용 증가
Tip 4: Early Stopping

engineer = LoopEngineer(max_passes=5)
engineer.convergence_threshold = 0.02 # 2%로 조정
``

결과적으로:

참고 자료

- Liu et al., "Lost in the Middle: How Language Models Use Long Contexts", 2023