1
┌─────────────┐     ┌─────────────────┐     ┌──────────────────┐
2
│  Feedback    │────▶│ Experimentation │────▶│   Continuous     │
3
│  Pipelines   │     │                 │     │   Learning       │
4
│              │     │ Shadow Deploy   │     │                  │
5
│ 관찰/진단/   │     │ A/B Testing     │     │ In-Context       │
6
│ 우선순위화   │◀────│ Bayesian Bandit │◀────│ Offline Retrain  │
7
└─────────────┘     └─────────────────┘     └──────────────────┘
8
     ▲                                              │
9
     └──────────────────────────────────────────────┘

1
from dataclasses import dataclass, field
2
from enum import Enum
3

4
class Phase(Enum):
5
    FEEDBACK = "feedback"
6
    EXPERIMENT = "experiment"
7
    LEARN = "learn"
8

9
@dataclass
10
class ImprovementLoop:
11
    """멀티에이전트 시스템의 피드백 기반 개선 사이클"""
12
    cycle: int = 0
13
    insights: list[str] = field(default_factory=list)
14
    validated: list[str] = field(default_factory=list)
15

16
    def feedback(self, failures: list[str]) -> list[str]:
17
        """1단계: 실패 관찰 → 패턴 클러스터링 → 우선순위화"""
18
        # 실제로는 자동 RCA + HITL 리뷰가 여기서 수행됨
19
        prioritized = sorted(failures, key=lambda f: failures.count(f), reverse=True)
20
        self.insights = list(dict.fromkeys(prioritized))  # 중복 제거, 빈도순 유지
21
        return self.insights
22

23
    def experiment(self, insight: str, improved: bool) -> bool:
24
        """2단계: Shadow deploy / A/B test로 개선안 검증"""
25
        if improved:
26
            self.validated.append(insight)
27
        return improved
28

29
    def learn(self, insight: str, method: str = "in_context") -> str:
30
        """3단계: 검증된 개선을 시스템에 내재화"""
31
        self.cycle += 1
32
        return f"[Cycle {self.cycle}] '{insight}' applied via {method}"
33

34
# --- 시뮬레이션 ---
35
loop = ImprovementLoop()
36

37
# 1) Feedback: 반복 실패 패턴 수집
38
raw_failures = [
39
    "tool_selection_error", "prompt_ambiguity",
40
    "tool_selection_error", "tool_selection_error",  # 빈도 높음
41
    "timeout", "prompt_ambiguity",
42
]
43
insights = loop.feedback(raw_failures)
44
print("우선순위화된 인사이트:", insights)
45
# → ['tool_selection_error', 'prompt_ambiguity', 'timeout']
46

47
# 2) Experiment: 가장 빈도 높은 이슈 검증
48
top_issue = insights[0]
49
is_valid = loop.experiment(top_issue, improved=True)
50
print(f"'{top_issue}' 개선 검증: {is_valid}")
51

52
# 3) Learn: 시스템 반영
53
result = loop.learn(top_issue, method="prompt_refinement")
54
print(result)
55
# → [Cycle 1] 'tool_selection_error' applied via prompt_refinement

1
Initial Prompt → Target Model → Output
2
                                   ↓
3
                            Evaluation Model → Score
4
                                   ↓
5
                          Optimization Model → New Prompt (반복)

1
from dataclasses import dataclass
2

3
@dataclass
4
class PromptCandidate:
5
    text: str
6
    score: float = 0.0
7

8
def evaluate_prompt(prompt: str, test_cases: list[dict]) -> float:
9
    """평가 모델 시뮬레이션: 프롬프트가 테스트 케이스에 얼마나 부합하는지 점수화"""
10
    score = 0.0
11
    for case in test_cases:
12
        # 실제로는 LLM 호출 → 출력 → 메트릭 비교
13
        if case["keyword"] in prompt:
14
            score += 1.0
15
    return score / len(test_cases)
16

17
def optimize_prompt(current: str, feedback: str) -> str:
18
    """최적화 모델 시뮬레이션: 피드백 기반으로 프롬프트 수정 제안"""
19
    # 실제로는 LLM이 피드백을 보고 새 프롬프트를 생성
20
    return f"{current}\nAdditional guidance: {feedback}"
21

22
# --- APO 루프 ---
23
test_cases = [
24
    {"input": "Suspicious IP login", "expected": "lookup threat intel", "keyword": "threat intel"},
25
    {"input": "Malware hash detected", "expected": "query logs", "keyword": "logs"},
26
    {"input": "Unusual outbound traffic", "expected": "triage incident", "keyword": "triage"},
27
]
28

29
prompt = "You are a SOC analyst. Investigate security alerts."
30
history: list[PromptCandidate] = []
31

32
for iteration in range(3):
33
    score = evaluate_prompt(prompt, test_cases)
34
    history.append(PromptCandidate(text=prompt, score=score))
35
    print(f"[Iter {iteration}] Score: {score:.2f}")
36

37
    if score >= 1.0:
38
        break
39

40
    # 점수가 낮은 케이스에서 피드백 생성
41
    missing = [c["keyword"] for c in test_cases if c["keyword"] not in prompt]
42
    feedback = f"Always consider: {', '.join(missing)}"
43
    prompt = optimize_prompt(prompt, feedback)
44

45
print(f"\n최종 프롬프트:\n{prompt}")
46
print(f"최종 점수: {history[-1].score:.2f} → {evaluate_prompt(prompt, test_cases):.2f}")

1
Workflow Tracing → Fault Localization → Pattern Recognition → Impact Assessment
2
(결정 체인 재구성)  (문제 컴포넌트 격리)  (일회성 vs 반복 패턴)    (빈도 × 심각도)

1
from collections import Counter
2
from dataclasses import dataclass
3
from enum import Enum
4

5
class Severity(Enum):
6
    LOW = 1
7
    MEDIUM = 2
8
    HIGH = 3
9

10
@dataclass
11
class FailureEvent:
12
    component: str      # 어떤 tool/skill에서 발생
13
    error_type: str     # 에러 분류
14
    trace_id: str       # 워크플로우 추적용
15

16
@dataclass
17
class RCAResult:
18
    component: str
19
    frequency: int
20
    severity: Severity
21
    root_cause: str
22
    priority_score: float  # frequency × severity
23

24
class AutomatedFeedbackPipeline:
25
    def __init__(self):
26
        self.events: list[FailureEvent] = []
27

28
    def ingest(self, events: list[FailureEvent]):
29
        self.events.extend(events)
30

31
    def detect_patterns(self) -> dict[str, int]:
32
        """규칙 기반 + 통계적 클러스터링으로 반복 패턴 탐지"""
33
        return Counter(
34
            (e.component, e.error_type) for e in self.events
35
        )
36

37
    def run_rca(self, severity_map: dict[str, Severity]) -> list[RCAResult]:
38
        """탐지된 패턴에 대해 RCA 수행 → 우선순위화"""
39
        patterns = self.detect_patterns()
40
        results = []
41
        for (component, error_type), freq in patterns.items():
42
            sev = severity_map.get(error_type, Severity.LOW)
43
            results.append(RCAResult(
44
                component=component,
45
                frequency=freq,
46
                severity=sev,
47
                root_cause=f"{component}에서 {error_type} 반복 발생",
48
                priority_score=freq * sev.value,
49
            ))
50
        # 우선순위 높은 순 정렬
51
        return sorted(results, key=lambda r: r.priority_score, reverse=True)
52

53
# --- 시뮬레이션 ---
54
pipeline = AutomatedFeedbackPipeline()
55

56
# SOC 에이전트의 실패 이벤트 수집
57
pipeline.ingest([
58
    FailureEvent("query_logs", "invalid_syntax", "trace-001"),
59
    FailureEvent("query_logs", "invalid_syntax", "trace-002"),
60
    FailureEvent("query_logs", "invalid_syntax", "trace-003"),
61
    FailureEvent("lookup_threat_intel", "timeout", "trace-004"),
62
    FailureEvent("triage_incident", "wrong_classification", "trace-005"),
63
    FailureEvent("triage_incident", "wrong_classification", "trace-006"),
64
])
65

66
severity_map = {
67
    "invalid_syntax": Severity.MEDIUM,
68
    "timeout": Severity.LOW,
69
    "wrong_classification": Severity.HIGH,
70
}
71

72
rca_results = pipeline.run_rca(severity_map)
73
for r in rca_results:
74
    print(f"[Priority {r.priority_score:.0f}] {r.root_cause} "
75
          f"(빈도:{r.frequency}, 심각도:{r.severity.name})")

1
[Priority 6] triage_incident에서 wrong_classification 반복 발생 (빈도:2, 심각도:HIGH)
2
[Priority 6] query_logs에서 invalid_syntax 반복 발생 (빈도:3, 심각도:MEDIUM)
3
[Priority 1] lookup_threat_intel에서 timeout 반복 발생 (빈도:1, 심각도:LOW)

1
Input Data → Agent → Generated Output Candidates
2
                              ↓
3
                      Human Reviewer ←── Manual Feedback
4
                              ↓
5
                    Human-Approved Output → End Users
6
                              ↓
7
                      System Feedback (루프 백)

1
from dataclasses import dataclass
2
from enum import Enum
3

4
class Decision(Enum):
5
    AUTO_APPROVE = "auto_approve"
6
    ESCALATE = "escalate"
7

8
@dataclass
9
class AgentOutput:
10
    incident_id: str
11
    response: str
12
    confidence: float          # 0~1, 모델 자체 평가
13
    severity: str              # "low", "medium", "high", "critical"
14
    affects_critical_asset: bool
15

16
SEVERITY_WEIGHT = {"low": 1, "medium": 2, "high": 3, "critical": 4}
17

18
def compute_risk_score(output: AgentOutput) -> float:
19
    """uncertainty × consequence 기반 리스크 스코어"""
20
    uncertainty = 1.0 - output.confidence
21
    consequence = SEVERITY_WEIGHT.get(output.severity, 1)
22
    if output.affects_critical_asset:
23
        consequence *= 1.5
24
    return uncertainty * consequence
25

26
def escalation_router(output: AgentOutput, threshold: float = 1.0) -> Decision:
27
    """리스크 스코어 기반 에스컬레이션 판단"""
28
    score = compute_risk_score(output)
29
    decision = Decision.ESCALATE if score > threshold else Decision.AUTO_APPROVE
30
    return decision
31

32
# --- 시뮬레이션 ---
33
outputs = [
34
    AgentOutput("INC-001", "False positive from VPN",
35
                confidence=0.92, severity="low", affects_critical_asset=False),
36
    AgentOutput("INC-002", "Possible data exfiltration",
37
                confidence=0.55, severity="critical", affects_critical_asset=True),
38
    AgentOutput("INC-003", "Suspicious login attempt",
39
                confidence=0.75, severity="medium", affects_critical_asset=False),
40
    AgentOutput("INC-004", "Malware hash detected on admin server",
41
                confidence=0.60, severity="high", affects_critical_asset=True),
42
]
43

44
for o in outputs:
45
    score = compute_risk_score(o)
46
    decision = escalation_router(o)
47
    print(f"[{o.incident_id}] risk={score:.2f} → {decision.value}"
48
          f"  (conf={o.confidence}, sev={o.severity})")

1
[INC-001] risk=0.08 → auto_approve  (conf=0.92, sev=low)
2
[INC-002] risk=2.70 → escalate      (conf=0.55, sev=critical)
3
[INC-003] risk=0.50 → auto_approve  (conf=0.75, sev=medium)
4
[INC-004] risk=1.80 → escalate      (conf=0.60, sev=high)

1
Contextual Analysis → Trace Inspection → Impact Assessment → Resolution Design
2
(통제 환경에서 재현)   (로그/트레이스 검토)  (범위·심각도 평가)    (수정안 설계)

1
from dataclasses import dataclass, field
2
from datetime import datetime
3

4
@dataclass
5
class HITLReview:
6
    incident_id: str
7
    reviewer: str
8
    role: str                          # "engineer", "data_scientist", "pm", "ux"
9
    finding: str
10
    resolution: str
11
    reviewed_at: datetime = field(default_factory=datetime.now)
12

13
@dataclass
14
class HITLKnowledgeBase:
15
    """리뷰 결과를 축적하여 조직 학습에 활용"""
16
    reviews: list[HITLReview] = field(default_factory=list)
17

18
    def add_review(self, review: HITLReview):
19
        self.reviews.append(review)
20

21
    def find_similar(self, keyword: str) -> list[HITLReview]:
22
        """과거 유사 케이스 검색 → 반복 문제 식별"""
23
        return [r for r in self.reviews if keyword in r.finding]
24

25
    def recurrence_report(self) -> dict[str, int]:
26
        """동일 resolution이 반복되면 시스템적 문제 시그널"""
27
        from collections import Counter
28
        return Counter(r.resolution for r in self.reviews)
29

30
# --- 시뮬레이션 ---
31
kb = HITLKnowledgeBase()
32

33
kb.add_review(HITLReview(
34
    "INC-002", "Alice", "engineer",
35
    finding="triage_incident가 credential stuffing을 IP brute-force로 오분류",
36
    resolution="프롬프트에 credential stuffing 예시 추가",
37
))
38
kb.add_review(HITLReview(
39
    "INC-004", "Bob", "data_scientist",
40
    finding="isolate_host가 critical ops 서버를 확인 없이 격리",
41
    resolution="isolate_host에 confirmation step 추가",
42
))
43
kb.add_review(HITLReview(
44
    "INC-007", "Carol", "engineer",
45
    finding="새로운 공격 벡터(API key 탈취)를 기존 프롬프트가 커버 못함",
46
    resolution="프롬프트에 credential stuffing 예시 추가",  # 같은 resolution 반복!
47
))
48

49
# 반복 resolution 확인 → 시스템적 문제 시그널
50
report = kb.recurrence_report()
51
for resolution, count in report.most_common():
52
    flag = "시스템적 문제!" if count > 1 else ""
53
    print(f"[{count}회] {resolution}{flag}")

1
[2회] 프롬프트에 credential stuffing 예시 추가 시스템적 문제!
2
[1회] isolate_host에 confirmation step 추가

1
Rewriting for Clarity  →  명확한 지시, 응답 포맷 명시
2
Adding Exemplars       →  positive/negative 예시로 추론 앵커링
3
Decomposing Tasks      →  복잡한 멀티스텝을 순차적 서브프롬프트로 분리
4
Context Expansion      →  추가 배경/제약/도메인 지식 주입

1
import dspy
2

3
# LM 설정
4
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
5

6
# 모의 툴 정의
7
def lookup_threat_intel(indicator: str) -> str:
8
    """위협 인텔리전스 조회"""
9
    return f"Intel for {indicator}: potentially malicious"
10

11
def query_logs(query: str) -> str:
12
    """보안 로그 검색"""
13
    return f"Logs for '{query}': suspicious activity detected"
14

15
# 학습 데이터: alert → 기대 response
16
trainset = [
17
    dspy.Example(
18
        alert="Suspicious login from IP 203.0.113.45 to admin account.",
19
        response="Lookup threat intel for IP, query auth logs, triage as true positive."
20
    ).with_inputs('alert'),
21
    dspy.Example(
22
        alert="Unusual file download from example.com/malware.exe.",
23
        response="Lookup intel for URL and hash, query endpoint logs, isolate host."
24
    ).with_inputs('alert'),
25
    dspy.Example(
26
        alert="Multiple failed logins from new device.",
27
        response="Query auth logs, lookup device IP intel, triage if attack pattern."
28
    ).with_inputs('alert'),
29
]
30

31
# ReAct 모듈: 추론 + 툴 호출을 자동으로 엮음
32
react = dspy.ReAct("alert -> response", tools=[lookup_threat_intel, query_logs])
33

34
# MIPROv2 옵티마이저: 프롬프트/few-shot을 자동 생성
35
optimizer = dspy.MIPROv2(
36
    metric=dspy.evaluate.answer_exact_match,
37
    auto="light",
38
    num_threads=24,
39
)
40
optimized_react = optimizer.compile(react, trainset=trainset)
41

42
# optimized_react는 내부 프롬프트가 자동 개선된 ReAct 모듈
43
# → 기존 SOC 에이전트 워크플로우에 drop-in 교체 가능

1
Refining Internal Logic    → 툴 내부의 프롬프트/모델 최적화
2
Expanding Capabilities     → 새로운 시나리오 커버를 위한 기능 확장
3
Integration Improvements   → 툴 간 입출력 호환성 보장

1
import dspy
2

3
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
4

5
# 위협 분류기의 Signature 정의
6
class ThreatClassifier(dspy.Signature):
7
    """Classify threat level of an indicator as 'benign', 'suspicious', or 'malicious'."""
8
    indicator: str = dspy.InputField(desc="IP, URL, or file hash")
9
    threat_level: str = dspy.OutputField(desc="benign, suspicious, or malicious")
10

11
# Chain of Thought으로 추론 과정을 거쳐 분류
12
class ThreatClassificationModule(dspy.Module):
13
    def __init__(self):
14
        super().__init__()
15
        self.classify = dspy.ChainOfThought(ThreatClassifier)
16

17
    def forward(self, indicator):
18
        return self.classify(indicator=indicator)
19

20
# 학습 데이터: indicator → 정답 threat_level
21
trainset = [
22
    dspy.Example(indicator="203.0.113.45", threat_level="suspicious").with_inputs('indicator'),
23
    dspy.Example(indicator="example.com/malware.exe", threat_level="malicious").with_inputs('indicator'),
24
    dspy.Example(indicator="benign-site.net", threat_level="benign").with_inputs('indicator'),
25
    dspy.Example(indicator="abc123def456", threat_level="malicious").with_inputs('indicator'),
26
    dspy.Example(indicator="192.168.1.1", threat_level="benign").with_inputs('indicator'),
27
    dspy.Example(indicator="obfuscated.url/with?params", threat_level="suspicious").with_inputs('indicator'),
28
]
29

30
# 정확도 메트릭
31
def threat_match(example, pred, trace=None):
32
    return example.threat_level.lower() == pred.threat_level.lower()
33

34
# BootstrapFewshot: 학습 데이터에서 few-shot 예시를 자동 선별하여 프롬프트에 삽입
35
optimizer = dspy.BootstrapFewshotWithRandomSearch(
36
    metric=threat_match,
37
    max_bootstrapped_demos=4,   # 자동 생성 예시 최대 수
38
    max_labeled_demos=4,        # 라벨된 예시 최대 수
39
)
40
optimized_classifier = optimizer.compile(
41
    ThreatClassificationModule(), trainset=trainset
42
)
43

44
# 최적화된 분류기를 tool 함수에 통합
45
def classify_threat(indicator: str) -> str:
46
    """최적화된 DSPy 모듈로 위협 수준 분류"""
47
    prediction = optimized_classifier(indicator=indicator)
48
    return prediction.threat_level