1
2
3
4
5
6
7
8
9

graph TD
    A[사용자 질문 + 전체 컨텍스트] --> B[Self-Attention 계산]
    B --> C{컨텍스트 길이}
    C -->|짧음 < 4K tokens| D[높은 Attention 집중도]
    C -->|중간 4K-16K tokens| E[점진적 정보 분산]
    C -->|김 > 16K tokens| F[Lost in the Middle 발생]
    D --> G[정확한 응답]
    E --> H[간헐적 환각]
    F --> I[핵심 맥락 상실 및 환각 증가]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

from dataclasses import dataclass, field
from typing import List, Optional
import tiktoken

@dataclass
class ContextNode:
    """컨텍스트의 개별 단위를 나타내는 클래스"""
    id: str
    role: str  # 'user', 'assistant', 'system', 'tool'
    content: str
    token_count: int
    timestamp: float
    importance: float = 0.0  # 0.0 ~ 1.0
    is_pinned: bool = False  # 사용자가 명시적으로 보존 지정
    tags: List[str] = field(default_factory=list)

class ContextManager:
    """Contexty 스타일의 컨텍스트 관리자 구현체"""
    
    def __init__(self, max_tokens: int = 128000, model: str = "gpt-4o"):
        self.nodes: List[ContextNode] = []
        self.max_tokens = max_tokens
        self.model = model
        self.encoding = tiktoken.encoding_for_model(model)
        
    def add_node(self, node: ContextNode) -> None:
        """새로운 컨텍스트 노드 추가"""
        node.token_count = len(self.encoding.encode(node.content))
        self.nodes.append(node)
        self._auto_manage_context()
        
    def pin_node(self, node_id: str) -> None:
        """특정 노드를 보존 목록에 고정"""
        for node in self.nodes:
            if node.id == node_id:
                node.is_pinned = True
                break
                
    def unpin_node(self, node_id: str) -> None:
        """고정 해제"""
        for node in self.nodes:
            if node.id == node_id:
                node.is_pinned = False
                break
    
    def get_active_context(self) -> List[ContextNode]:
        """현재 활성화된 컨텍스트 반환 (PIN된 것 + 최근 것들)"""
        pinned = [n for n in self.nodes if n.is_pinned]
        unpinned = [n for n in self.nodes if not n.is_pinned]
        
        # PIN되지 않은 노드 중 최근 것들을 토큰 한계 내에서 포함
        available_tokens = self.max_tokens - sum(n.token_count for n in pinned)
        
        active_unpinned = []
        for node in reversed(unpinned):  # 가장 최근 것부터
            if available_tokens >= node.token_count:
                active_unpinned.insert(0, node)
                available_tokens -= node.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

token_count
            else:
                break
                
        return pinned + active_unpinned
    
    def get_context_stats(self) -> dict:
        """컨텍스트 통계 정보 반환"""
        pinned_nodes = [n for n in self.nodes if n.is_pinned]
        active_nodes = self.get_active_context()
        
        return {
            "total_nodes": len(self.nodes),
            "pinned_nodes": len(pinned_nodes),
            "active_nodes": len(active_nodes),
            "total_tokens": sum(n.token_count for n in self.nodes),
            "active_tokens": sum(n.token_count for n in active_nodes),
            "token_utilization": sum(n.token_count for n in active_nodes) / self.max_tokens * 100
        }
        
    def _auto_manage_context(self) -> None:
        """내부적으로 토큰 한계를 초과하면 오래된 비핀 노드 제거"""
        while True:
            active = self.get_active_context()
            active_tokens = sum(n.token_count for n in active)
            if active_tokens <= self.max_tokens:
                break
            # 가장 오래된 비핀 노드 찾아서 제거
            for i, node in enumerate(self.nodes):
                if not node.is_pinned:
                    self.nodes.pop(i)
                    break
            else:
                break  # 모든 노드가 PIN된 상태

# 사용 예시
if __name__ == "__main__":
    manager = ContextManager(max_tokens=8000)
    
    # 초기 아키텍처 결정 (PIN 필요)
    arch_node = ContextNode(
        id="arch_001",
        role="user",
        content="우리 프로젝트는 Clean Architecture를 사용하며, "
                "의존성 주입은 Hilt를 사용합니다. "
                "Repository 패턴을 적용하고 UseCase 계층을 분리합니다.",
        token_count=0,
        timestamp=1000.0,
        importance=1.0,
        tags=["architecture", "convention"]
    )
    manager.add_node(arch_node)
    manager.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

pin_node("arch_001")  # 핵심 맥락 고정!
    
    # 일반 디버깅 대화 (여러 턴)
    for i in range(15):
        debug_node = ContextNode(
            id=f"debug_{i:03d}",
            role="user",
            content=f"이 오류 로그를 확인해줘: NullPointerException at line {i*10}",
            token_count=0,
            timestamp=1000.0 + i * 100,
        )
        manager.add_node(debug_node)
    
    # 통계 확인
    stats = manager.get_context_stats()
    print(f"활성 노드: {stats['active_nodes']} / {stats['total_nodes']}")
    print(f"토큰 사용률: {stats['token_utilization']:.1f}%")
    print(f"고정된 노드: {stats['pinned_nodes']}")