Embedding과 벡터 검색

Embedding 모델의 작동 원리

Embedding 모델은 텍스트를 고차원 벡터로 변환하여, 의미적으로 유사한 텍스트가 벡터 공간에서 가까운 거리에 위치하게 합니다.

"나는 사과 먹는 것을 좋아한다"   →  [0.12, -0.34, 0.56, ..., 0.78]  (1024차원)
"나는 사과 먹는 것을 사랑한다"   →  [0.11, -0.33, 0.55, ..., 0.79]  ← 거리가 매우 가까움
"자동차 수리 가이드"             →  [-0.78, 0.45, -0.23, ..., 0.01] ← 거리가 매우 멈

Embedding 모델의 학습은 일반적으로 대조 학습(Contrastive Learning) 단계를 포함합니다: 입력은 관련성/비관련성이 레이블된 많은 텍스트 쌍이며, 학습 목표는 관련 텍스트의 벡터 유사도를 최대화하고 비관련 텍스트의 유사도를 최소화하는 것입니다.

# 두 텍스트 벡터의 코사인 유사도 계산
import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 예시
query_vec = embedding_model.encode("연차는 어떻게 신청하나요?")
doc_vec = embedding_model.encode("직원 연차 신청 절차")

similarity = cosine_similarity(query_vec, doc_vec)
print(f"유사도: {similarity:.4f}")  # 0.92 — 높은 관련성

벡터 데이터베이스 선정

벡터 데이터베이스는 RAG 시스템의 핵심 인프라입니다. 선택 시 다음을 고려해야 합니다:

# Chroma 사용 예시 (경량 로컬 벡터 DB)
import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection(
    name="company_docs",
    embedding_function=embedding_functions.OpenAIEmbeddingFunction(
        api_key="your-api-key",
        model_name="text-embedding-3-small"
    )
)

# 문서 추가
collection.add(
    documents=["직원 연차 신청 절차...", "출장 정산 기준..."],
    metadatas=[{"source": "hr_policy.pdf"}, {"source": "finance_policy.pdf"}],
    ids=["doc_1", "doc_2"]
)

# 검색
results = collection.query(
    query_texts=["연차는 어떻게 신청하나요?"],
    n_results=3
)

문서 청크 분할 전략

분할의 기본 모순

RAG 시스템의 검색 효과는 문서 청크 분할 품질에 크게 의존합니다. 분할의 핵심 모순은:

청크가 너무 큼 → 검색 시 너무 많은 노이즈 도입, 모델 주의력 분산
청크가 너무 작음 → 핵심 정보가 잘릴 수 있음, 컨텍스트 손실

“만능” 최적 분할 전략은 존재하지 않습니다. 문서 유형, 검색 시나리오, 모델 능력에 따라 선택해야 합니다.

다섯 가지 주요 분할 방법

Token 분할

고정 token 수로 분할하며, token 소비를 정밀하게 제어해야 하는 시나리오에 적합합니다.

from llama_index.core.node_parser import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=256,     # 각 청크의 token 수
    chunk_overlap=30    # 인접 청크 간 중복 token 수
)

nodes = splitter.get_nodes_from_documents(documents)

장점: 컨텍스트 크기를 정밀하게 제어, 작은 컨텍스트 윈도우 모델에 적합. 단점: 문장 중간에서 잘릴 수 있어 의미적 완전성을 해칠 수 있음.

문장 분할

문장의 완전성을 유지하는 분할 방식으로, 대부분의 시나리오에서 기본 선택입니다.

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50
)

장점: 자연어 의미 단위의 완전성 유지. 단점: 문서 구조에 대한 인식 부족, 관련 단락을 다른 청크로 분할할 수 있음.

문장 윈도우 청크

인덱싱과 검색에 서로 다른 세분화 사용: 인덱싱은 작은 세분화로 정밀 매칭, 검색 반환 시 인접 컨텍스트 윈도우 포함.

from llama_index.core.node_parser import SentenceWindowNodeParser

parser = SentenceWindowNodeParser(
    window_size=3,          # 검색 시 3개의 인접 문장 확장
    window_metadata_key="window",
    original_text_metadata_key="original"
)

핵심 장점: 검색 정밀도와 컨텍스트 완전성을 모두 고려.

의미적 청크

의미적 관련성에 따라 적응적으로 분할 지점을 선택하여 문서의 의미적 연속성 유지.

from llama_index.core.node_parser import SemanticSplitterNodeParser

splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,  # 유사도가 이 임계값 이하일 때 분할
    embed_model=embed_model
)

적용 시나리오: 논리성이 좋고 내용이 전문적인 긴 문서.

Markdown 청크

Markdown 구조화 문서에 특화된 최적화, 제목 계층에 따라 분할.

from llama_index.core.node_parser import MarkdownNodeParser

parser = MarkdownNodeParser()
# #, ##, ### 등의 제목 계층을 자동 인식하여 각 제목 단락에서 분할

모범 사례: 문서를 PDF/Word에서 Markdown으로 변환한 후 분할하여, 제목 구조를 활용해 검색 정확도 향상.

분할 전략 선택 가이드

검색 증강 생성 파이프라인

RAG의 2단계 아키텍처

RAG(Retrieval-Augmented Generation)는 LLM의 “지식 부족”을 해결하는 핵심 아키텍처입니다. 프로세스를 두 단계로 나눕니다:

1단계: 인덱스 구축

1. 문서 파싱: PDF, Word, Markdown 등의 형식을 순수 텍스트로 파싱
2. 텍스트 청크: 선택한 전략에 따라 문서를 단락으로 분할
3. 벡터화: Embedding 모델로 각 청크를 벡터로 변환
4. 인덱스 저장: 벡터를 벡터 데이터베이스에 저장하고 인덱스 구축

2단계: 검색과 생성

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ 사용자 질문 │ → │ 벡터 검색  │ → │ Prompt 조립 │ → │ 모델 생성  │
│ Query     │    │ Retrieve  │    │ Augment   │    │ Generate  │
└──────────┘    └──────────┘    └──────────┘    └──────────┘

1. 사용자 질문: 사용자 질문 수신
2. 벡터 검색: 질문을 벡터화하여 벡터 DB에서 가장 유사한 청크 검색
3. Prompt 조립: 검색된 지식 조각 + 원본 질문 + 지시사항을 완전한 Prompt로 조립
4. 모델 생성: LLM이 증강된 컨텍스트를 기반으로 답변 생성

완전한 RAG 파이프라인 예시

from openai import OpenAI
import numpy as np

client = OpenAI()

class SimpleRAG:
    def __init__(self, embed_model="text-embedding-3-small"):
        self.embed_model = embed_model
        self.documents = []      # 문서 텍스트 저장
        self.embeddings = []     # 문서 벡터 저장

    def add_documents(self, docs: list[str]):
        """인덱스 구축: 문서를 벡터화하여 저장"""
        for doc in docs:
            vec = self._embed(doc)
            self.documents.append(doc)
            self.embeddings.append(vec)

    def _embed(self, text: str) -> np.ndarray:
        resp = client.embeddings.create(
            model=self.embed_model,
            input=text
        )
        return np.array(resp.data[0].embedding)

    def retrieve(self, query: str, top_k: int = 3) -> list[str]:
        """가장 관련성 높은 문서 조각 검색"""
        query_vec = self._embed(query)
        similarities = [
            np.dot(query_vec, doc_vec) /
            (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec))
            for doc_vec in self.embeddings
        ]
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices]

    def query(self, question: str) -> str:
        """완전한 RAG 쿼리"""
        contexts = self.retrieve(question)
        prompt = f"""다음 참조 정보에 따라 질문에 답변해주세요:

참조 정보:
{' '.join(contexts)}

질문: {question}

참조 정보가 질문에 답변하기에 부족하면 명확히 밝혀주세요."""

        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        return resp.choices[0].message.content

RAG 멀티턴 대화에서의 질문 재작성

RAG 시나리오에서 멀티턴 대화를 구현하는 것은 독특한 도전 과제입니다. 사용자가 두 번째 턴에 “그의 상사는 누구인가요?”라고 묻는다면, 이 문장만으로 검색하면 완전히 실패할 것입니다 — 시스템은 “그”가 누구를 가리키는지 모릅니다.

해결책: 질문 재작성(Query Rewriting)

def rewrite_query(conversation_history: list, current_query: str) -> str:
    """LLM을 사용하여 컨텍스트 의존적 질문을 독립적 질문으로 재작성"""
    rewrite_prompt = f"""
    대화 기록에 따라 현재 질문을 컨텍스트에 의존하지 않는 독립적 질문으로 재작성해주세요.

    대화 기록:
    {format_history(conversation_history)}

    현재 질문: {current_query}

    재작성된 질문:"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": rewrite_prompt}],
        temperature=0.1
    )
    return resp.choices[0].message.content

# 예시
# 기록: 사용자 "장삼의 자리는 어디인가요?" 어시스턴트 "A동 5층입니다"
# 현재: "그의 상사는 누구인가요?"
# 재작성 후: "장삼의 상사는 누구인가요?"

고급 RAG 패턴

HyDE(Hypothetical Document Embeddings)

HyDE의 핵심 아이디어: 먼저 모델이 가상의 답변을 “만들어내게” 한 다음, 그 가상 답변으로 검색하고 원래 질문은 사용하지 않는 것입니다. 이 접근법의 직관은 — 가상 답변이 질문보다 의미적으로 실제 문서에 더 가깝다는 것입니다.

def hyde_retrieve(query: str, top_k: int = 3) -> list[str]:
    """HyDE 방법을 사용한 검색"""
    # Step 1: 가상 답변 생성
    hyde_prompt = f"""
    Question: {query}
    Please write a passage that answers this question.
    Passage:"""

    hyde_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": hyde_prompt}]
    )
    hypothetical_doc = hyde_resp.choices[0].message.content

    # Step 2: 가상 답변으로 원래 질문 대신 검색
    query_vec = embed(hypothetical_doc)
    results = vector_db.search(query_vec, top_k=top_k)
    return results

핵심 원칙:

• 알고리즘을 최적화하기 전에 먼저 부족한 지식을 보충하십시오
• 재현율을 개선하기 전에 먼저 문서 품질을 향상시키십시오
• 지속적으로 사용자 의도를 수집하여 “데이터 수집-지식 업데이트-전문가 검증” 피드백 루프 형성

Function Calling 프로토콜

AutoGen Studio — Microsoft의 노코드 멀티 에이전트 워크플로우 빌더 — microsoft/autogen 제공

DSPy — LLM을(프롬프트가 아닌)선언적으로 프로그래밍;Hermes 자가 진화 파이프라인이 사용 — stanfordnlp/dspy 제공

Function Calling이란

Function Calling(함수 호출, Tool Calling이라고도 함)은 LLM API가 제공하는 표준 기능입니다. 모델이 필요할 때 순수 텍스트 응답 대신 구조화된 도구 호출 명령을 출력할 수 있도록 합니다.

워크플로우는 다음과 같습니다:

JSON Schema 도구 정의

# 도구 목록 정의
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "회사 내부 지식 베이스를 검색하여 정책 문서와 운영 가이드를 가져옵니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 키워드 또는 질문"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["hr", "it", "finance", "general"],
                        "description": "지식 카테고리"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "이메일 전송",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {
                        "type": "string",
                        "description": "수신자 이메일"
                    },
                    "subject": {
                        "type": "string",
                        "description": "이메일 제목"
                    },
                    "body": {
                        "type": "string",
                        "description": "이메일 본문"
                    }
                },
                "required": ["to", "subject", "body"]
            }
        }
    }
]

완전한 Function Calling 루프

from openai import OpenAI
import json

client = OpenAI()

def execute_function_call(tool_call):
    """도구 호출을 실행하고 결과 반환"""
    func_name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    if func_name == "search_knowledge_base":
        # 실제 검색 로직
        result = knowledge_base.search(args["query"])
        return json.dumps(result)
    elif func_name == "send_email":
        # 실제 메일 전송 로직
        result = email_service.send(
            to=args["to"],
            subject=args["subject"],
            body=args["body"]
        )
        return json.dumps({"status": "sent" if result else "failed"})
    else:
        return json.dumps({"error": f"Unknown function: {func_name}"})

def chat_with_tools(user_message: str, messages: list = None):
    """Function Calling을 지원하는 대화"""
    if messages is None:
        messages = [
            {"role": "system", "content": "당신은 기업 어시스턴트로, 지식 베이스 검색과 이메일 전송이 가능합니다."}
        ]

    messages.append({"role": "user", "content": user_message})

    # 첫 번째 호출: 모델이 도구 사용 여부 결정
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools
    )

    assistant_msg = response.choices[0].message

    # 모델이 도구를 호출하려는 경우
    if assistant_msg.tool_calls:
        messages.append(assistant_msg)

        for tool_call in assistant_msg.tool_calls:
            # 도구 실행
            result = execute_function_call(tool_call)
            # 결과를 모델에 반환
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # 두 번째 호출: 모델이 도구 결과를 기반으로 최종 응답 생성
        final_response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        return final_response.choices[0].message.content

    # 모델이 직접 응답한 경우
    return assistant_msg.content

도구 정의 모범 사례

1. 설명은 정확하게: 모델은 description에 따라 도구 호출 시점을 판단하므로, 설명이 모호하면 오호출로 이어집니다
2. 파라미터에 제약 적용: enum, required, 타입 제약을 사용하여 모델의 파라미터 오류 감소
3. 함수 책임은 단일하게: 하나의 함수에 여러 작업을 포함하지 마십시오
4. 구조화된 결과 반환: 도구 반환값은 모델이 이해하기 쉽도록 JSON 사용 권장

# ❌ 좋지 않은 도구 설명
{"name": "do_stuff", "description": "작업 실행", "parameters": {...}}

# ✅ 정밀한 도구 설명
{
    "name": "cancel_meeting",
    "description": "지정된 회의를 취소합니다. 회의 ID와 취소 사유가 필요합니다",
    "parameters": {
        "properties": {
            "meeting_id": {"type": "string", "description": "회의의 고유 식별자"},
            "reason": {"type": "string", "description": "취소 사유, 모든 참석자에게 통지됩니다"}
        },
        "required": ["meeting_id"]
    }
}

ReAct 추론-행동 루프

ReAct의 핵심 아이디어

ReAct(Reasoning + Acting)는 모델이 **추론(Thought)**과 **행동(Action)**을 번갈아 수행하는 패턴입니다. 한 번에 최종 답변을 생성하는 것이 아니라 “사고→행동→관찰→사고…”의 순환을 통해 문제를 해결합니다.

전형적인 ReAct 실행 과정:

사용자: "장삼의 부서를 확인하고 그의 상사에게 이메일을 보내주세요"

Thought 1: 먼저 장삼의 부서 정보를 찾아야 합니다
Action 1: search_knowledge_base(query="장삼 부서")
Observation 1: "장삼은 교연부 소속, 상사는 리시([email protected])"

Thought 2: 정보를 얻었으니, 이제 리시에게 이메일을 작성해야 합니다
Action 2: send_email(to="[email protected]", subject="장삼 관련",
                      body="...")
Observation 2: {"status": "sent"}

Thought 3: 작업 완료
Final Answer: "장삼이 교연부 소속임을 확인했으며, 그의 상사 리시에게 이메일을 발송했습니다."

수동 ReAct Agent 구현

class ReActAgent:
    def __init__(self, tools: dict, max_iterations: int = 10):
        self.tools = tools
        self.max_iterations = max_iterations

    def run(self, task: str) -> str:
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": task}
        ]

        for i in range(self.max_iterations):
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=self._format_tools()
            )

            msg = response.choices[0].message

            if msg.content and not msg.tool_calls:
                # 모델이 최종 답변을 제시함
                return msg.content

            if msg.tool_calls:
                # 어시스턴트의 도구 호출을 기록에 추가
                messages.append(msg)

                for tc in msg.tool_calls:
                    tool_name = tc.function.name
                    args = json.loads(tc.function.arguments)

                    # 도구 실행
                    result = self.tools[tool_name](**args)

                    # 관찰 결과를 기록에 추가
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": json.dumps(result)
                    })
                    print(f"  [Tool: {tool_name}({args}) → {result}]")

        return "ReAct 루프가 최대 반복 횟수에 도달했습니다"

    def _build_system_prompt(self) -> str:
        return """당신은 도구를 사용할 수 있는 지능형 어시스턴트입니다.
ReAct 패턴을 따릅니다: 먼저 사고하고, 행동하고, 결과를 관찰한 후, 다음 단계를 결정합니다.
작업이 완료되면 최종 답변을 직접 제시합니다."""

    def _format_tools(self) -> list:
        return [
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": func.__doc__ or "",
                    "parameters": get_schema(func)
                }
            }
            for name, func in self.tools.items()
        ]

ReAct의 장점과 한계

장점:

• 관찰 가능: 각 단계의 사고와 행동을 추적할 수 있음
• 오류 수정 가능: 잘못된 결과를 관찰한 후 전략 조정 가능
• 조합 가능: 여러 도구를 자동으로 솔루션으로 조합

한계:

• 루프 횟수 예측 불가 (무한 루프 가능성)
• 여러 번의 API 호출로 지연 시간과 비용 증가
• 도구가 반환하는 관찰 품질에 의존

MCP 프로토콜과 도구 생태계

왜 MCP가 필요한가

Function Calling에는 근본적인 문제가 있습니다: 도구 정의와 소비가 결합되어 있다는 점입니다. 모든 Agent 개발자는 자신의 코드에 도구의 JSON Schema를 하드코딩해야 합니다. 도구 API가 업그레이드되면, 해당 도구를 통합한 모든 Agent가 수동으로 업데이트해야 합니다.

전통적 Function Calling 패턴:
  Agent A ──하드코딩 Schema──→ web_search v1
  Agent B ──하드코딩 Schema──→ web_search v1  ← 중복 정의
  Agent C ──하드코딩 Schema──→ web_search v1  ← 중복 정의

MCP 패턴:
  Agent A ──┐
  Agent B ──┼── MCP Client ──→ MCP Server (web_search)
  Agent C ──┘                    ↑
                          도구 제공자가 Schema 정의

**MCP(Model Context Protocol)**의 핵심 아이디어는 “도구를 제공하는 자가 도구를 정의한다”입니다. 도구 정의의 책임을 Agent(소비자 측)에서 도구 서비스(제공자 측)로 이전합니다.

MCP의 아키텍처 역할

MCP Server 및 Client 구축

MCP Server 예시:

from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent

app = Server("web-search")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="web_search",
            description="인터넷에서 최신 정보 검색",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "검색 키워드"},
                    "num_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "web_search":
        results = search_engine.search(
            arguments["query"],
            num=arguments.get("num_results", 5)
        )
        return [TextContent(type="text", text=json.dumps(results))]
    raise ValueError(f"Unknown tool: {name}")

# stdio로 Server 시작
async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

MCP Client 통합 예시:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_with_mcp_tools(user_query: str):
    server_params = StdioServerParameters(
        command="python",
        args=["web_search_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # MCP Server의 도구 정의 가져오기
            tools_result = await session.list_tools()
            tools = tools_result.tools

            # OpenAI 형식의 tools 파라미터로 변환
            openai_tools = [
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                }
                for tool in tools
            ]

            # 표준 Function Calling 프로세스
            response = client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": user_query}],
                tools=openai_tools
            )

            # 모델이 도구 호출을 원하면 MCP Client를 통해 실행
            if response.choices[0].message.tool_calls:
                for tc in response.choices[0].message.tool_calls:
                    result = await session.call_tool(
                        tc.function.name,
                        json.loads(tc.function.arguments)
                    )
                    # ... 결과를 모델에 반환

MCP의 엔지니어링 가치

• 디커플링: 도구 정의와 서비스 구현이 분리되어 각자 독립적 반복 가능
• 동적 발견: Agent 시작 시 자동으로 최신 도구 목록 가져오기, 유지보수 비용 제로
• 생태계 효과: 서드파티가 표준화된 MCP Server를 제공할 수 있으며, Agent 개발자는 MCP Client만 통합하면 됨
• 다중 전송 프로토콜: stdio(로컬 프로세스 통신) 및 HTTP/SSE(원격 통신) 지원

성찰과 자기 교정

왜 성찰이 필요한가

LLM이 생성한 내용이 항상 사용 가능한 것은 아닙니다. 다음과 같은 문제가 발생할 수 있습니다:

• 코드의 변수명을 “조용히 수정”하여 실행 오류 발생
• 잘못된 “사실”에 기반하여 추론을 계속하여 연쇄 오류 발생
• 긴 출력에서 앞부분의 제약 조건을 잊어버림

성찰(Reflection)의 핵심 아이디어는: 모델이 자신이 이미 생성한 완전한 내용을 검토하고 평가할 기회를 주어, 오류를 발견하고 수정하게 하는 것입니다.

자기 피드백의 두 가지 패턴

패턴 1: 단일 단계 명령식 성찰

한 번의 호출로, Prompt를 통해 모델이 답변 생성과 동시에 성찰을 수행하도록 지시합니다:

prompt_with_reflection = """
## 작업
1. 다음 강좌의 언어 표현을 다듬고, 다듬은 전체 내용을 출력하세요.
2. 다듬은 내용을 성찰하세요:
   - 작성 규범을 준수하는가
   - 언어 표현 외에 다른 내용이 의도치 않게 수정되었는가
   성찰 결과와 수정 제안을 출력하세요.
3. 제안에 따라 강좌를 수정하고, 수정된 전체 내용을 출력하세요.

## 강좌 초안
{original_content}
"""

장점: 구현이 간단하고 한 번의 호출로 완료. 단점: 모델이 동일한 사고 편향으로 자기 검증하기 쉬워 “자기 정당화”에 빠질 수 있음.

패턴 2: 2단계 “생성-피드백”

생성과 심사를 두 번의 독립적 호출로 분리합니다:

def generate_and_review(content: str) -> str:
    # 1단계: 생성
    draft_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "당신은 강좌 작가입니다. 다음 내용을 더 매력적으로 다듬으세요."
        }, {
            "role": "user",
            "content": content
        }]
    )
    draft = draft_resp.choices[0].message.content

    # 2단계: 심사 (다른 system prompt 사용!)
    review_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": """당신은 엄격한 기술 심사관입니다.
[원본 내용]과 [다듬은 내용]을 비교하세요:
- 문장 표현만 수정되고 코드 등 기술 내용이 완전히 동일 → "통과"라고 답변
- 기술 내용이 수정됨 → "불통과"라고 답변하고 구체적 위치 지적"""
        }, {
            "role": "user",
            "content": f"원본 내용:\n{content}\n\n다듬은 내용:\n{draft}"
        }]
    )

    # 3단계: 불통과 시 심사 결과를 작성 Agent에게 피드백하여 수정
    review = review_resp.choices[0].message.content
    if "불통과" in review:
        # ... 피드백을 작성 Agent에게 전달하여 수정
        pass

    return draft

핵심 장점: 심사 Agent의 관점이 작성 Agent와 달라 역할 편향을 회피합니다. 나아가 여러 전문 심사 Agent — 사실 심사, 논리 심사, 스타일 심사, 보안 심사 — 를 설정할 수도 있습니다.

외부 피드백

자기 피드백에는 본질적 한계가 있습니다: 모델은 내용이 실제 환경에서 올바른지 검증할 수 없습니다. 외부 피드백의 아이디어는 생성 결과를 실제 환경에서 실행하여 객관적 사실로 검증하는 것입니다.

def generate_code_and_validate(spec: str) -> str:
    # 1단계: 코드 생성
    code_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"다음 요구사항에 따라 Python 코드를 작성하세요:\n{spec}"}]
    )
    code = extract_code(code_resp.choices[0].message.content)

    # 2단계: 외부 실행 검증
    import subprocess, tempfile
    with tempfile.NamedTemporaryFile(suffix=".py", mode="w") as f:
        f.write(code)
        f.flush()
        result = subprocess.run(
            ["python", f.name],
            capture_output=True,
            text=True,
            timeout=30
        )

    # 3단계: 오류를 모델에 피드백하여 수정
    if result.returncode != 0:
        fix_prompt = f"""다음 코드 실행 중 오류 발생:

코드:
{code}

오류 정보:
{result.stderr}

코드를 수정하고 완전한 수정 버전을 출력하세요."""
        fix_resp = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": fix_prompt}]
        )
        return fix_resp.choices[0].message.content

    return code

외부 피드백의 적용 시나리오:

• 코드 실행 검증: 코드 인터프리터로 코드를 실행하여 런타임 오류 캡처
• JSON Schema 검증: Pydantic 등의 라이브러리로 구조화된 출력 검증
• 수치 계산 검증: 계산기 도구로 수학 결과 검증
• 시각화 렌더링 검증: 차트 생성 후 모델이 렌더링 결과를 “보고” 시각적 검사

Plan & Execute 패턴

왜 명시적 계획이 필요한가

Agent가 복잡한 작업을 직접 실행할 때 흔히 발생하는 문제:

• 망각: 뒤쪽 단계를 처리할 때 앞쪽의 제약 조건을 “잊어버림”
• 연쇄 오류: 초기 오류가 이후 추론의 기반이 되어 오류가 기하급수적으로 증폭
• 구조 누락: 모델이 선형 처리를 선호하여 작업의 병렬/의존 관계를 인식하지 못함

Plan & Execute 패턴의 핵심 아이디어: 먼저 계획하고 실행하라 — 먼저 완전한 행동 계획을 수립하고, 검토 확인 후 단계적으로 실행하는 것입니다.

Plan Mode 구현

from typing import List
from pydantic import BaseModel

class PlanStep(BaseModel):
    step_id: int
    description: str
    dependencies: List[int] = []  # 의존하는 단계 ID
    tool: str = ""               # 사용할 도구
    expected_output: str = ""    # 예상 산출물

class ExecutionPlan(BaseModel):
    goal: str
    steps: List[PlanStep]

def plan_and_execute(task: str) -> str:
    # Phase 1: 계획 수립
    plan_prompt = f"""
    당신은 프로젝트 계획 전문가입니다. 다음 작업에 대한 상세 실행 계획을 수립하세요.

    요구사항:
    1. 작업을 구체적 단계로 분해
    2. 단계 간 의존 관계 표시
    3. 각 단계의 예상 산출물 설명

    작업: {task}

    JSON 형식으로 계획을 출력하세요."""

    plan_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": plan_prompt}],
        response_format={"type": "json_object"}
    )
    plan = ExecutionPlan.model_validate_json(
        plan_resp.choices[0].message.content
    )

    # Phase 2: 의존 관계에 따라 실행
    results = {}
    executed = set()

    while len(executed) < len(plan.steps):
        for step in plan.steps:
            if step.step_id in executed:
                continue
            # 의존성이 모두 실행되었는지 확인
            if all(dep in executed for dep in step.dependencies):
                # 단계 실행
                result = execute_step(step, results)
                results[step.step_id] = result
                executed.add(step.step_id)

    # Phase 3: 결과 취합
    return summarize_results(plan, results)

고정 워크플로우: Pipeline 패턴

작업의 단계가 확정적이고 반복 가능할 때는 파이프라인으로 고정해야 합니다:

Input → Step 1 → Step 2 → Step 3 → ... → Output

class Pipeline:
    """고정 파이프라인: 각 단계의 출력이 다음 단계의 입력"""

    def __init__(self):
        self.steps = []

    def add_step(self, name: str, func):
        self.steps.append({"name": name, "func": func})

    def run(self, input_data):
        result = input_data
        for step in self.steps:
            print(f"  [실행] {step['name']}")
            result = step["func"](result)
        return result

# 예시: 문서 처리 파이프라인
pipeline = Pipeline()
pipeline.add_step("PDF 파싱", parse_pdf_to_text)
pipeline.add_step("텍스트 정제", clean_text)
pipeline.add_step("분할", split_sections)
pipeline.add_step("벡터화", vectorize_chunks)
pipeline.add_step("인덱스 저장", store_to_vectordb)

pipeline.run("document.pdf")

워크플로우 오케스트레이션 패턴

다섯 가지 핵심 워크플로우 패턴

복잡한 작업은 Agent 노드를 특정 토폴로지로 구성해야 합니다. 다음은 다섯 가지 핵심 패턴입니다:

분기 선택(Branching/Router)

진입 노드에서 작업 유형을 판단하여 서로 다른 처리 경로로 분배합니다.

def router_agent(user_input: str):
    """의도에 따라 다른 처리 파이프라인으로 분배"""
    classify_prompt = f"""
    다음 사용자 요청의 유형을 분석하고 한 단어만 응답하세요:
    - code_review: 코드 검사/검증
    - style_review: 언어 교정/최적화
    - fact_check: 사실/개념 정확성 검증

    요청: {user_input}
    유형:"""

    intent = client.chat.completions.create(
        model="gpt-4o-mini",  # 경량 모델로 비용 절감
        messages=[{"role": "user", "content": classify_prompt}],
        temperature=0
    ).choices[0].message.content.strip()

    pipelines = {
        "code_review": code_review_pipeline,
        "style_review": style_review_pipeline,
        "fact_check": fact_check_pipeline
    }
    return pipelines.get(intent, default_pipeline)(user_input)

병렬 실행(Parallel)

서로 의존하지 않는 하위 작업을 동시에 분배하고 마지막에 취합합니다.

              ┌→ 코드 검사 ──┐
사용자 입력 → Split ─┼→ 사실 심사 ──┼→ Merge → 종합 보고서
              └→ 스타일 심사 ─┘

import asyncio

async def parallel_review(notebook_content: str):
    """강좌 콘텐츠에 대해 세 가지 심사 병렬 수행"""
    tasks = [
        asyncio.create_task(check_code(notebook_content)),
        asyncio.create_task(check_facts(notebook_content)),
        asyncio.create_task(check_style(notebook_content))
    ]

    code_result, fact_result, style_result = await asyncio.gather(*tasks)

    # 취합
    return generate_summary_report(code_result, fact_result, style_result)

혼합 전문가(Mixture-of-Agents, MoA)

여러 다른 모델이 동일한 작업을 처리하고, 집계자가 최적의 결과를 종합합니다.

             ┌→ 모델 A (추론 특화) ──┐
사용자 질문 → Split ─┼→ 모델 B (창의 특화) ──┼→ Aggregator → 최적 답변
             └→ 모델 C (정확 특화) ─┘

MoA의 핵심 발견은 모델의 “협업성”(Collaborativeness)입니다: 한 모델이 다른 모델의 출력을 참조할 수 있을 때, 종종 더 높은 품질의 응답을 생성할 수 있습니다.

def mixture_of_agents(task: str):
    """MOA 구현: 다중 모델 + 집계"""
    # 1계층: 제안자 병렬 생성
    proposers = ["gpt-4", "claude-3-opus", "gemini-pro"]
    proposals = []

    for model in proposers:
        resp = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": task}]
        )
        proposals.append(resp.choices[0].message.content)

    # 2계층: 집계자 종합
    aggregator_prompt = f"""
    다음은 동일한 질문에 대한 {len(proposals)}개의 답변입니다. 각 답변의 장점을 종합하여
    최적의 답변을 생성하세요.

    {format_proposals(proposals)}

    종합 답변:"""

    final = client.chat.completions.create(
        model="gpt-4",  # 가장 강력한 모델로 집계
        messages=[{"role": "user", "content": aggregator_prompt}]
    )
    return final.choices[0].message.content

휴먼-인-더-루프(Human-in-the-Loop, HITL)

중요 지점에서 인간 검토를 도입하여 “AI 실행 → 인간 승인 → AI 계속” 순환을 형성합니다.

def hitl_workflow(task: str):
    """휴먼-인-더-루프 워크플로우"""
    plan = generate_plan(task)
    print(f"실행 계획:\n{format_plan(plan)}")

    approval = input("이 계획을 승인하시겠습니까? (y/n): ")
    if approval.lower() != 'y':
        return "작업이 취소되었습니다"

    for step in plan.steps:
        result = execute_step(step)
        print(f"단계 {step.step_id} 완료: {result['summary']}")

        if step.get("requires_review"):
            review = input(f"단계 결과를 검토하세요 (approve/modify/reject): ")
            if review == "reject":
                print("단계가 거부되었습니다. 재실행 중...")
                result = execute_step(step, feedback=review)

    return generate_final_output()

패턴 선택 방법론

모범 사례는 “탐색-고정” 혼합 모드입니다: 먼저 Plan & Execute로 최적 방안을 탐색한 후, Pipeline으로 고정하여 대규모 생산에 사용합니다.

계층적 협업 패턴

Leader-Worker 아키텍처

계층적 협업(Hierarchical/Team Leader Pattern)은 가장 직관적인 다중 Agent 협업 패턴입니다. “프로젝트 매니저 + 팀원”의 조직 구조를 모방합니다:

Leader Agent의 책임:

1. 최상위 작업 수신 및 이해
2. 하위 작업으로 분해하여 적합한 Worker에게 할당
3. 전체 진행 상황 추적
4. Worker 성과 취합

Worker Agent는 각자 특정 분야의 전문성을 보유하며, 할당된 하위 작업 실행에 집중합니다.

Handoff를 통한 계층적 협업 구현

class LeaderWorkerSystem:
    """계층적 협업 시스템 구현"""

    def __init__(self):
        self.workers = {
            "instructional_designer": self._create_worker(
                "당신은 교육 설계자로, 강좌 개요와 학습 경로 설계를 전문으로 합니다."
            ),
            "data_scientist": self._create_worker(
                "당신은 데이터 과학자로, Python 데이터 분석 코드와 사례 작성을 전문으로 합니다."
            ),
            "content_writer": self._create_worker(
                "당신은 콘텐츠 작성자로, 기술 내용을 생생한 강좌 원고로 변환하는 것을 전문으로 합니다."
            )
        }
        self.leader = self._create_leader()

    def _create_leader(self):
        return {
            "system_prompt": """당신은 강좌 프로젝트 책임자입니다. 당신의 책임은:
1. 요구사항 분석, 하위 작업으로 분해
2. 적합한 전문가에게 작업 할당
3. 각 전문가의 성과를 완전한 강좌로 통합
사용 가능한 전문가 팀: instructional_designer, data_scientist, content_writer""",
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "delegate_to_worker",
                        "description": "지정된 전문가에게 하위 작업 할당",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "worker": {
                                    "type": "string",
                                    "enum": list(self.workers.keys()),
                                    "description": "작업을 받을 전문가"
                                },
                                "task": {
                                    "type": "string",
                                    "description": "구체적인 하위 작업 설명"
                                }
                            },
                            "required": ["worker", "task"]
                        }
                    }
                }
            ]
        }

    def run(self, project_brief: str) -> str:
        """완전한 강좌 개발 프로젝트 실행"""
        messages = [
            {"role": "system", "content": self.leader["system_prompt"]},
            {"role": "user", "content": project_brief}
        ]

        # Leader 루프
        while True:
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=self.leader["tools"]
            )
            msg = response.choices[0].message

            if msg.content and not msg.tool_calls:
                return msg.content  # 최종 취합 출력

            if msg.tool_calls:
                messages.append(msg)
                for tc in msg.tool_calls:
                    if tc.function.name == "delegate_to_worker":
                        args = json.loads(tc.function.arguments)
                        # Worker를 호출하여 하위 작업 실행
                        worker_result = self._run_worker(
                            args["worker"], args["task"]
                        )
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tc.id,
                            "content": worker_result
                        })

    def _run_worker(self, worker_name: str, task: str) -> str:
        worker = self.workers[worker_name]
        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": worker},
                {"role": "user", "content": task}
            ]
        )
        return resp.choices[0].message.content

계층적 협업의 장단점

장점:

• 구조가 명확하고 각 Agent의 책임이 분명
• Leader가 전체를 통제하므로 목표 이탈 없음
• 각 Worker가 독립적인 컨텍스트 윈도우를 보유하여 더 집중
• 병렬 작업 할당 지원

단점:

• Worker 간 직접 소통 불가, 정보 전달에 지연/왜곡 발생
• Leader가 단일 병목 지점
• 각 모듈 조합 후 전체적 유려함 부족 가능성

블랙보드 협업 패턴

탈중앙화된 공동 창작

블랙보드 패턴(Blackboard/Co-creation Pattern)은 “전문가들이 화이트보드 앞에 둘러앉아 브레인스토밍하는” 작업 방식을 모방합니다. 중앙화된 조정자가 없으며, 모든 Agent가 평등하게 공유 공간을 읽고 씁니다:

블랙보드 패턴 구현

class BlackboardSystem:
    """블랙보드 협업 시스템"""

    def __init__(self, agents: dict, max_rounds: int = 3):
        self.agents = agents
        self.max_rounds = max_rounds
        self.blackboard = []  # 공유 공간

    def run(self, problem: str) -> str:
        # 문제를 블랙보드에 작성
        self.blackboard.append({"source": "user", "content": problem})

        for round_num in range(self.max_rounds):
            print(f"\n=== {round_num + 1} 라운드 ===")
            new_contributions = []

            # 모든 Agent가 병렬로 블랙보드를 읽고 기여
            for name, agent_config in self.agents.items():
                contribution = self._agent_contribute(
                    name, agent_config, self.blackboard
                )
                if contribution:
                    new_contributions.append({
                        "source": name,
                        "content": contribution
                    })

            # 새로운 기여를 블랙보드에 추가
            self.blackboard.extend(new_contributions)

            # 합의 도달 확인
            if self._check_consensus():
                break

        return self._synthesize_final_answer()

    def _agent_contribute(self, name, config, blackboard):
        """각 Agent가 블랙보드를 읽고 자신의 아이디어 기여"""
        board_text = self._format_blackboard(blackboard)

        prompt = f"""당신은 {config['role']}입니다.

현재 공유 블랙보드의 내용:
{board_text}

기존 논의를 바탕으로 당신의 견해, 보충, 질문 또는 새로운 아이디어를 제시하세요.
이미 제안된 방안이 완벽하다면 동의하고 그 이유를 설명하셔도 됩니다."""

        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": config["system_prompt"]},
                {"role": "user", "content": prompt}
            ]
        )
        return resp.choices[0].message.content

    def _format_blackboard(self, blackboard):
        return "\n\n".join([
            f"[{entry['source']}]: {entry['content']}"
            for entry in blackboard
        ])

    def _check_consensus(self):
        """블랙보드상의 최신 기여가 합의를 형성했는지 확인"""
        # 합의 감지 로직 구현
        pass

    def _synthesize_final_answer(self):
        """블랙보드 내용에서 최종 방안 합성"""
        pass

블랙보드 패턴 vs 계층적 패턴

협업 패턴 선택 방법론

현실 세계에서 배우기

훌륭한 다중 Agent 시스템 설계는 현실 세계 팀 협업에 대한 관찰과 정제에서 비롯됩니다. 추상적인 패턴 이름을 암기하기보다, 비즈니스 현장에 들어가 인간 전문가 팀이 유사한 작업을 어떻게 완수하는지 관찰하십시오.

관찰의 세 가지 차원:

혼합 패턴 설계

실제 프로젝트에서 단일 패턴만 사용하는 경우는 드뭅니다. 일반적인 것은 혼합 설계입니다:

                    ┌─────────────┐
                    │   Leader    │  ← 계층적 패턴
                    └──────┬──────┘
           ┌───────────────┼───────────────┐
           ▼               ▼               ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ 교육 설계자 │    │ 콘텐츠 작성자 │    │ 심사 Leader│
    │  Worker   │    │  Worker   │    └─────┬────┘
    └──────────┘    └──────────┘           │
                                  ┌────────┼────────┐
                                  ▼        ▼        ▼
                             ┌──────┐ ┌──────┐ ┌──────┐
                             │코드 검│ │사실 검│ │스타일│  ← 병렬 패턴
                             └──────┘ └──────┘ └──────┘
                                 │        │        │
                                 └────────┼────────┘
                                          ▼
                                    ┌──────────┐
                                    │ 종합 보고서 │
                                    └──────────┘

비용 의식

다중 Agent 시스템의 Token 소비는 일반적으로 단일 Agent의 3-5배입니다. 설계 시 다음을 고려해야 합니다:

def estimate_cost(num_agents: int, avg_tokens_per_agent: int,
                  rounds: int = 1, price_per_1k: float = 0.01):
    """다중 Agent 시스템의 Token 비용 추정"""
    total_tokens = num_agents * avg_tokens_per_agent * rounds
    return total_tokens * price_per_1k / 1000

# 예시: 5개 Agent, 각 2000 tokens, 3 라운드 블랙보드 논의
cost = estimate_cost(5, 2000, 3)
print(f"추정 비용: ${cost:.2f}")
# 실제 수치는 블랙보드 내용의 반복 전송으로 인해 더 높을 수 있음

단기 메모리 관리

무상태성: 문제의 근원

LLM은 본질적으로 **무상태(Stateless)**입니다. 각 API 호출은 독립적입니다 — 이전 대화 내용, 당신의 선호도, 또는 이전에 도달한 합의를 기억하지 않습니다.

# 이 두 호출은 서로 완전히 독립적
response1 = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "제 이름은 장삼입니다"}]
)
# 응답: "안녕하세요 장삼님! 무엇을 도와드릴까요?"

response2 = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "제 이름이 뭐죠?"}]
)
# 응답: "죄송합니다, 이전 대화가 없어서 당신의 이름을 알 수 없습니다."

해결책: 대화 기록 목록을 유지하고, 호출할 때마다 전체 기록을 전송합니다.

class ConversationBuffer:
    """가장 간단한 단기 메모리: 완전한 대화 기록 저장"""

    def __init__(self, system_prompt: str = ""):
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})

    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        response = client.chat.completions.create(
            model="gpt-4",
            messages=self.messages
        )
        reply = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})
        return reply

컨텍스트 윈도우 압박

대화 턴이 증가함에 따라 전체 기록 방식은 세 가지 치명적인 문제에 직면합니다:

1. 컨텍스트 윈도우 초과: 기록 길이가 모델 제한 초과 → 프로그램 오류
2. 비용 통제 불능: 호출할 때마다 전체 기록 재전송 → Token 소비 선형 증가
3. 주의력 분산: 긴 컨텍스트에서 모델의 중간 부분 정보 처리 능력이 현저히 저하

세 가지 메모리 관리 전략

전략 1: 고정 윈도우 절단(Context Truncation)

최근 N 턴의 대화 또는 N 개의 Token만 유지합니다.

class TruncationMemory:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.messages = []

    def add_and_truncate(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # 가장 오래된 메시지부터 삭제하여 총 token 수가 제한 내에 들도록 함
        while self._total_tokens() > self.max_tokens:
            self.messages.pop(0)  # 가장 오래된 non-system 메시지 삭제

    def _total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

장점: 구현이 매우 간단하고 계산 오버헤드가 작음. 단점: 핵심 정보가 초기 대화에 있으면 절단 후 Agent가 “기억 상실”.

전략 2: 롤링 요약(Rolling Summary)

잊기 전에 먼저 핵심을 추출합니다.

대화 기록: [msg1, msg2, msg3, msg4, msg5, msg6, msg7, msg8]
                          ↓ 앞부분 압축
         [요약(m1-m4), msg5, msg6, msg7, msg8]
                          ↓ 계속 압축
         [요약(m1-m6), msg7, msg8]

class RollingSummaryMemory:
    def __init__(self, summary_trigger_tokens: int = 3000):
        self.summary_trigger = summary_trigger_tokens
        self.messages = []
        self.summary = ""

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        if self._total_tokens() > self.summary_trigger:
            self._compress()

    def _compress(self):
        """앞부분 대화를 요약으로 압축"""
        split_point = len(self.messages) // 2
        to_compress = self.messages[:split_point]
        remaining = self.messages[split_point:]

        compress_prompt = f"""
        다음 대화 기록을 간결한 요약으로 정리하고 핵심 정보를 유지하세요:

        대화:
        {format_messages(to_compress)}

        요약:"""

        resp = client.chat.completions.create(
            model="gpt-4o-mini",  # 경량 모델로 요약
            messages=[{"role": "user", "content": compress_prompt}]
        )
        self.summary = resp.choices[0].message.content

        # 요약으로 압축된 메시지 대체
        self.messages = [
            {"role": "system", "content": f"대화 기록 요약:\n{self.summary}"}
        ] + remaining

    def _total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

장점: 길이를 압축하면서도 핵심 정보를 보존하여 장기 일관성 유지. 단점: 추가 API 호출 비용, 요약 품질이 후속 대화에 직접적 영향.

전략 3: 벡터화 검색(Vector-based Retrieval)

가장 지능적인 방식: 대화 기록을 벡터 데이터베이스에 저장하고 필요에 따라 가장 관련성 높은 기억을 검색합니다.

class VectorBasedMemory:
    def __init__(self):
        self.conversations = []  # 완전한 대화 기록
        self.embeddings = []     # 각 턴 대화의 벡터
        self.embed_model = "text-embedding-3-small"

    def store_conversation(self, user_msg: str, assistant_msg: str):
        """한 턴의 대화를 저장하고 벡터화"""
        conversation_text = f"User: {user_msg}\nAssistant: {assistant_msg}"
        self.conversations.append(conversation_text)

        vec = client.embeddings.create(
            model=self.embed_model,
            input=conversation_text
        )
        self.embeddings.append(vec.data[0].embedding)

    def retrieve_relevant(self, current_query: str, top_k: int = 5):
        """현재 질문과 가장 관련성 높은 과거 대화 검색"""
        query_vec = client.embeddings.create(
            model=self.embed_model,
            input=current_query
        ).data[0].embedding

        # 유사도 계산
        similarities = [
            np.dot(query_vec, mem_vec) /
            (np.linalg.norm(query_vec) * np.linalg.norm(mem_vec))
            for mem_vec in self.embeddings
        ]

        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.conversations[i] for i in top_indices]

장점: 컨텍스트 윈도우의 길이 제한에서 근본적으로 해방, 의미적 정밀 매칭. 단점: 시스템 복잡도가 가장 높음, Embedding 모델과 벡터 DB 도입.

전략 선택 가이드

장기 메모리와 벡터 저장소

수동적 컨텍스트에서 능동적 메모리 관리로

진정한 지능형 Agent는 단순히 처리된 컨텍스트를 수동적으로 받아들이는 것을 넘어, 자신의 메모리를 능동적으로 관리할 수 있어야 합니다 — 언제 무엇을 기억해야 하는지, 언제 무엇을 떠올려야 하는지 스스로 결정하는 것입니다.

이를 위해 Agent에 두 가지 핵심 도구를 제공해야 합니다:

# Agent가 사용 가능한 메모리 관리 도구
memory_tools = [
    {
        "type": "function",
        "function": {
            "name": "record_to_memory",
            "description": "중요한 정보를 장기 메모리에 기록합니다. 사용자가 명시적으로 선호도를 표현하거나, 핵심 정보를 제공하거나, 중요한 결정을 내렸을 때 사용합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "content": {
                        "type": "string",
                        "description": "기억해야 할 내용"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["preference", "fact", "decision", "context"],
                        "description": "메모리 카테고리"
                    }
                },
                "required": ["content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "retrieve_from_memory",
            "description": "장기 메모리에서 관련 정보를 검색합니다. 사용자 선호도, 과거 결정 또는 이전에 논의된 내용을 회상해야 할 때 사용합니다.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 쿼리"
                    },
                    "category": {
                        "type": "string",
                        "description": "선택사항, 검색할 메모리 카테고리 제한"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

단기 메모리 vs 장기 메모리

┌─────────────────────────────────────────────────────┐
│                    메모리 시스템 아키텍처                │
├─────────────────────────────────────────────────────┤
│                                                     │
│  단기 메모리 (Short-term Memory)                       │
│  ┌───────────────────────────────────────────────┐  │
│  │ 저장: 대화 버퍼 (messages list)                │  │
│  │ 관리: 절단 / 요약                              │  │
│  │ 생명주기: 현재 세션                            │  │
│  │ 책임: 대화 일관성 유지, "방금 무엇을 이야기했는지" 기억 │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  장기 메모리 (Long-term Memory)                        │
│  ┌───────────────────────────────────────────────┐  │
│  │ 저장: 벡터 DB + 메타데이터 저장소               │  │
│  │ 관리: 벡터화 검색 + 도구화 호출                 │  │
│  │ 생명주기: 세션 간                               │  │
│  │ 책임: 핵심 정보 영속화, 세션 간 지능형 검색 지원  │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
└─────────────────────────────────────────────────────┘

메모리 관리 모범 사례

1. 선별적으로 기억하기: 기억은 많을수록 좋은 것이 아닙니다. 낮은 가치의 정보는 후속 검색을 방해합니다. 쓰기 진입 메커니즘을 구축하십시오 — 사용자가 명시적으로 요청하거나 정보의 중요도가 임계값을 초과할 때만 기록합니다.

2. 지속적 거버넌스: 메모리는 동적 데이터 자산입니다. 정기적으로 오래된 정보를 정리하고, 중복 항목을 병합하며, 사실 정확성을 검증하십시오. 사용자가 메모리를 관리할 수 있는 인터페이스(조회, 수정, 삭제)를 제공하십시오.

3. 시나리오별 적용: 시나리오마다 메모리 요구사항이 다릅니다. 강좌 문서 워크플로우에서는 개인화 선호도를 기록해서는 안 됩니다. 제품 사실 정보(API 파라미터, 기능 제한 등)에 대해서는 기록하고 정기적으로 유효성을 검토해야 합니다.

Skill 시스템 설계

Prompt에서 Skill로의 진화

정교하게 작성된 Prompt는 가치가 높지만, 현재 세션에서만 효력을 발휘합니다. 세션이 끝나면 지식은 흩어집니다. Skill은 Prompt를 재사용 가능하고, 버전 관리 가능하며, 팀 공유가 가능한 전문 지식 모듈로 업그레이드하는 것입니다.

진화 경로:

임시 Prompt ("이 강좌를 검토해 주세요")
    ↓ 문제: 매번 다시 작성해야 하고, 기준이 일관되지 않음
고정 Prompt (채팅 기록에 저장)
    ↓ 문제: 흩어져 있고, 찾기 어렵고, 협업 불가
독립 파일 (course-review.md)
    ↓ 문제: 파일이 비대해지고 유지보수 어려움
지식 베이스 디렉토리 (course-review/)
    ↓ 문제: 여전히 Agent에게 무엇을 해야 하는지 수동으로 알려줘야 함
Skill (SKILL.md + 리소스 파일 + 스크립트)

Skill의 구조

규범적인 Skill은 YAML frontmatter + Markdown 본문으로 구성됩니다:

---
name: course-review
description: |
  강좌 콘텐츠의 기술적 정확성, 코드 정확성 및 교육 품질을 검토합니다.
  사용자가 기존 강좌나 교육 자료의 검토, 감사 또는 평가를 요청할 때 이 스킬을 사용합니다.
---

# 강좌 검토 스킬

## 검토 프로세스
1. Notebook 디렉토리 구조 추출, 전체 챕터 구성 이해
2. 챕터별 단락별 검사, 다음 방향에 따라:
   - 코드 실행 가능성 (자세한 내용은 [code-quality.md](code-quality.md) 참조)
   - 콘텐츠 정확성 (자세한 내용은 [content-accuracy.md](content-accuracy.md) 참조)
   - 설명 스타일 (자세한 내용은 [style-guide.md](style-guide.md) 참조)
   - 구식 API (자세한 내용은 [outdated-api.md](outdated-api.md) 참조)
3. 검토 결과 취합, 출력 형식에 따라 보고서 생성

## 안티패턴 체크리스트
- ❌ 코드의 변수명, API 버전 번호를 수정하지 마십시오
- ❌ 어떤 검사 항목도 건너뛰지 마십시오
- ❌ 검토 과정에서 새로운 기술 개념을 도입하지 마십시오

## 출력 형식
각 검사 항목에 대해: 통과/불통과/인간 검토 필요 + 위치 + 수정 제안

디렉토리 구조:

course-review/
├── SKILL.md              # 메인 명령 진입점
├── code-quality.md       # 코드 실행 가능성 검사 항목
├── content-accuracy.md   # 사실 정확성 검사 항목
├── style-guide.md        # 설명 스타일 좋은 예/나쁜 예
├── outdated-api.md       # 구식 API 대조표
└── scripts/
    ├── extract_toc.py    # Notebook 디렉토리 추출
    └── validate_code.py  # 코드 검증 자동 실행

Skill vs RAG

많은 사람이 Skill과 RAG를 혼동합니다. 이들의 핵심 차이는:

고품질 Skill 작성을 위한 5단계

1단계: 할 가치가 있는지 판단
  ├─ 이 작업에 "전문가 직관"이 있는가? (전문가는 잘하지만 초보자는 경계 조건을 놓치기 쉬운)
  ├─ 이 작업이 충분히 복잡한가? (3단계 이내 GUI로 완료할 수 있으면 하지 않음)
  └─ 이 작업이 반복적으로 실행되는가? (한 번만 하면 하지 않음)

2단계: 무엇을 작성할지 추출
  ├─ 단순 단계가 아닌 전문가의 의사결정 트리 추출
  ├─ 안티패턴 검사 주입 ("절대 밟아서는 안 될 함정들")
  ├─ Template 패턴: 표준화된 출력 템플릿 제공
  └─ Examples 패턴: 예시로 문자 설명 대체

3단계: 명령 잘 작성하기
  ├─ 간결: 모든 문장이 token 비용만큼 가치 있어야 함
  ├─ 자유도 매칭: 제약 정도가 작업 리스크와 일치해야 함
  │   ├─ 낮은 자유도 (데이터베이스 마이그레이션): 정밀 스크립트
  │   ├─ 중간 자유도 (보고서 생성): 의사 코드/파라미터화
  │   └─ 높은 자유도 (코드 리뷰): 텍스트 명령
  └─ 점진적 공개: 메인 파일은 간결하게 유지, 세부사항은 필요에 따라 로드

4단계: 도구 갖추기
  ├─ 워크플로우: 추적 가능한 Checklist
  ├─ 피드백 루프: 실행→확인→수정→반복
  ├─ 고위험 작업: 먼저 계획 검증 후 실행
  └─ AI 친화적 스크립트: 구조화된 상태 + 수정 단서 + 우아한 저하 + 멱등 안전

5단계: 검증과 반복
  ├─ 단계 1: 평가 기준선 구축 (먼저 평가를 갖춘 후 Skill 작성)
  ├─ 단계 2: Skill 추출 (AI로 반복 제공되는 정보를 요약)
  └─ 단계 3: 이중 Agent 테스트 반복 (설계자 vs 사용자)

Skill as Code

Skill을 코드로 간주하고 코드 엔지니어링의 방법론을 적용합니다:

• 버전 관리: Skill 디렉토리를 Git에 포함, 모든 수정 사항은 이력 기록
• 코드 리뷰: 신규/수정 Skill은 PR 리뷰 필요
• CI/CD: Skill 변경 시 평가 파이프라인 트리거, 퇴행 방지
• 커뮤니티 공유: 오픈소스 라이브러리처럼 Skill 공유 및 재사용

점진적 정보 공개와 Skill as Code

점진적 정보 공개의 설계 철학

전통적인 Prompt 엔지니어링이 직면한 핵심 모순: 정보가 너무 많으면 → 컨텍스트 혼잡, 주의력 분산; 정보가 너무 적으면 → Agent가 의사결정을 뒷받침할 충분한 지식 부족.

**점진적 정보 공개(Progressive Disclosure)**는 이 모순을 해결하는 설계 패턴입니다:

제1계층: Skill 목록 (Agent 시작 시 표시)
  ↓ Agent가 특정 Skill 활성화 선택
제2계층: SKILL.md (Skill 활성화 후 로드)
  ↓ Agent가 특정 단계 실행
제3계층: 하위 파일 (필요에 따라 해당 리소스 로드)
  ↓ 특정 기술 세부사항 검증 필요
제4계층: 스크립트 실행 (결정론적 결과 제공)

설계 원칙:

• 평면 구조 유지, 깊은 중첩 참조 방지
• SKILL.md가 모든 리소스 파일을 직접 링크, “한 단계”로 도달 가능하게 함
• 모든 내용을 SKILL.md에 밀어 넣지 말 것 — 이는 단지 진입점일 뿐입니다

AI 친화적 스크립트 설계

도구 스크립트의 출력 품질은 Agent의 성능에 직접적 영향을 미칩니다. 네 가지 핵심 원칙:

1. 구조화된 상태 피드백 — 자유 텍스트 대신 JSON 출력:

# ❌ 좋지 않은 출력
print("Error: exit code 1")
print("Could not find module 'openpyxl'")

# ✅ AI 친화적 출력
print(json.dumps({
    "status": "failed",
    "error_code": "MODULE_NOT_FOUND",
    "missing_module": "openpyxl",
    "fix_hint": "pip install openpyxl을 실행하여 누락된 의존성 설치",
    "fallback_available": True,
    "affected_cells": [42, 43, 44]
}))

2. 오류 정보에 수정 단서 포함 — “무엇이 잘못되었는지”뿐 아니라 “어떻게 고칠지”를 Agent에게 알려줍니다.

3. 우아한 저하이(가) 아닌 충돌 — 가능한 경우 기본값을 제공하고 계속 진행합니다.

4. 멱등과 안전 — 부작용 없이 반복 실행 지원:

평가 프레임워크 설계

왜 “느낌이 괜찮다”는 믿을 수 없는가

Agent 최적화가 주관적 느낌에만 의존하면 다음과 같은 문제가 발생합니다:

• 정량화 어려움: “느낌이 더 좋다”는 엔지니어링 의사결정의 근거가 될 수 없음
• 기준 부재: 테스터와 시점에 따라 평가 기준이 표류할 수 있음
• 재현 불가: 체계적 회귀 테스트가 불가능하여 새 변경이 기존 기능을 망가뜨리지 않았는지 확인할 수 없음

**평가 주도 개발(Evaluation-Driven Development)**은 평가를 개발 프로세스의 끝에서 핵심 위치로 끌어올립니다:

         ┌──────────────────┐
         │  평가 = 품질 척도   │
         └────────┬─────────┘
                  │
    ┌─────────────┼─────────────┐
    ▼             ▼             ▼
측정 가능해야    피드백이 빠르고   제품 능력의
개선 가능하다    정확할수록       상한을 결정한다
                 개선이 효율적이다

엔드 투 엔드 평가(End-to-End Evaluation)

엔드 투 엔드 평가는 최종 출력에 주목하여 “이 Agent가 사용자에게 좋은가?”라는 질문에 답합니다.

평가 지표는 두 가지 유형으로 나뉩니다:

화이트박스 평가(White-box Evaluation)

Agent 프로세스가 복잡해지면 엔드 투 엔드 평가로는 구체적 문제를 찾을 수 없습니다. 화이트박스 평가는 시스템 내부로 깊이 들어가 주요 구성 요소별로 독립적 평가 체계를 설계할 것을 주장합니다.

엔드 투 엔드 평가 (최종 출력만 봄):
  Agent 전체 → 점수 4.2/5 → 하지만 어디가 문제인지 모름

화이트박스 평가 (중간 단계 검사):
  ┌────────────┐    ┌────────────┐    ┌────────────┐
  │ 개념 설명    │    │ 코드 생성    │    │ 스타일 조정   │
  │ 점수 3.1/5 │    │ 점수 4.8/5 │    │ 점수 4.5/5 │
  └────────────┘    └────────────┘    └────────────┘
       ↑ 병목이 여기!

화이트박스 평가의 핵심 장점:

• 명확한 신호: 간섭 없는 개선 신호, 진정한 병목에 집중
• 빠른 반복: 단일 구성 요소만 테스트, 전체 프로세스 실행 불필요
• 정밀 최적화: 각 변경 효과를 정확하게 측정 가능

LLM-as-Judge

LLM을 평가자로 사용하기

또 다른 LLM이 “평가 전문가” 역할을 하여, 정의된 지표와 채점 기준에 따라 자동으로 점수를 매깁니다.

def llm_judge_evaluate(generated_output: str, criteria: dict):
    """LLM을 평가자로 사용"""
    judge_prompt = f"""
    당신은 전문 강좌 품질 평가자입니다. 다음 기준에 따라 점수를 매기세요:

    평가 기준:
    {json.dumps(criteria, indent=2, ensure_ascii=False)}

    평가 대상 콘텐츠:
    {generated_output}

    JSON 형식으로 평가 결과를 출력하세요:
    {{
        "scores": {{
            "accuracy": <1-5>,
            "clarity": <1-5>,
            "engagement": <1-5>
        }},
        "overall": <1-5>,
        "comments": "종합 평가"
    }}"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": judge_prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(resp.choices[0].message.content)

LLM 평가자의 편향

LLM을 평가자로 사용할 때는 그 고유의 편향을 반드시 경계해야 합니다:

모범 사례: 초기에는 인간 전문가를 사용하여 “골든 테스트 세트”를 구축하고, 이를 사용해 LLM 평가자를 교정하십시오. 정기적으로 수동 샘플 검사로 자동 평가의 일관성을 검증하십시오.

평가 지표 분해

모호한 평가 목표를 하나씩 점검 가능한 구체적 세부 기준으로 분해합니다:

# 콘텐츠 품질 평가 분해 예시
평가 차원: "콘텐츠 품질"
세부 기준:
  - id: "pain_point"
    description: "구체적인痛点으로 시작하는가?"
    type: "boolean"
  - id: "theory_depth"
    description: "초기 해결책의 한계를 명확히 지적하고 핵심 이론으로 이끄는가?"
    type: "boolean"
  - id: "code_relevance"
    description: "코드 예시가 설명하는 이론과 밀접하게 관련되고 충분히 단순화되었는가?"
    type: "boolean"
  - id: "anti_pattern_check"
    description: "'새로운 스킬을 획득한 것을 축하합니다' 같은 농담을 피했는가?"
    type: "boolean"

평가 주도 반복

평가 폐쇄 루프

평가는 일회성이 아니라 지속적으로 개선을 이끄는 엔진입니다:

    ┌──────────────────────────────────┐
    │                                  │
    ▼                                  │
┌─────────┐   ┌──────────┐   ┌─────────┐
│ MVP 구축  │ → │ 문제 발견  │ → │ 지표 정제 │
└─────────┘   └──────────┘   └─────────┘
                                  │
                                  ▼
┌─────────┐   ┌──────────┐   ┌─────────┐
│ 배포/출시  │ ← │ 회귀 테스트 │ ← │ 최적화 개선 │
└─────────┘   └──────────┘   └─────────┘
                                  │
                                  └────→ (순환)

비즈니스 전문가 주도 평가 표준

평가 지표(특히 주관적 지표)는 반드시 가장 경험이 많은 비즈니스 전문가가 주도하여 수립해야 합니다:

1. 비즈니스 목표로 참여 유도: “평가 지표 정의를 도와주세요”가 아니라 “이 Agent는 당신의 강좌 제작 주기를 2주에서 3일로 단축하고, 90% 이상의 사용자 만족도를 유지해야 합니다”라고 말하십시오.

2. 구조화된 도구로 진입 장벽 낮추기: 채점 척도 템플릿, 사례 주석 도구, “강좌의 좋고 나쁨을 판단하기 위해 단 세 가지 지표만 볼 수 있다면 무엇을 선택하시겠습니까?”와 같은 유도 질문.

3. 지속적 협업 메커니즘 구축: 주간 리뷰 회의에서 전문가가 데이터를 보고, 기술팀이 시스템을 조정하며, 함께 의사결정합니다.

평가의 효율 레버리지

모든 평가가 전자동, 전체 커버리지일 필요는 없습니다. 가장 간단한 방법부터 시작하십시오:

Level 1: 수동 샘플 검사 → "코드를 복사해서 실행해 보기"
Level 2: 자동화 스크립트 → "스크립트를 작성하여 일괄 실행"
Level 3: 평가 파이프라인 통합 → CI/CD 통합, PR마다 자동 트리거
Level 4: 지속적 모니터링 → 프로덕션 환경 실시간 핵심 지표 모니터링

각 Level의 투자 대비 수익은 다릅니다. 초기에는 Level 1의 투자 대비 수익이 가장 높습니다 — 가장 빠르게 문제를 발견하고 구현 비용이 가장 낮습니다. 시스템이 성숙해짐에 따라 점진적으로 더 높은 Level로 진화하십시오.

모델 배포 전략

비즈니스 요구사항 분석 프레임워크

LLM 애플리케이션을 프로덕션 환경에 배포할 때, 첫 단계는 기술 선정이 아니라 요구사항 분석입니다:

┌────────────────────────────────────────────┐
│           비즈니스 요구사항 분석 매트릭스         │
├────────────────────────────────────────────┤
│                                            │
│  기능적 요구사항 (무엇을 하는가):                 │
│  ├─ 자연어 처리 → 범용 LLM                     │
│  ├─ 코드 생성 → 코드 최적화된 LLM               │
│  ├─ 수학 추론 → 수학 파인튜닝된 LLM             │
│  ├─ 시각 이해 → 멀티모달 모델                    │
│  └─ 음성 처리 → 음성 모델                       │
│                                            │
│  비기능적 요구사항 (어떻게 하는가):                │
│  ├─ 성능: TTFT < 500ms, TPOT < 50ms         │
│  ├─ 비용: 단일 호출 < $0.01                    │
│  ├─ 안정성: 99.9% 가용성                       │
│  ├─ 보안: 콘텐츠 필터링, 프라이버시 보호           │
│  └─ 컴플라이언스: 산업 규제 요구사항               │
│                                            │
└────────────────────────────────────────────┘

모델 선정 전략

모든 시나리오에 가장 큰 모델이 필요한 것은 아닙니다. 모델 선정은 “최소 가용” 원칙을 따릅니다:

작업 복잡도
    ↑
    │  ┌──────────────────────────┐
    │  │ 대형 모델 (GPT-4, Claude)   │
    │  │ - 복잡한 추론               │
    │  │ - 다단계 계획               │
    │  │ - 창의적 생성               │
    │  └──────────────────────────┘
    │  ┌──────────────────────────┐
    │  │ 중형 모델 (GPT-4o-mini)     │
    │  │ - 의도 인식                │
    │  │ - 구조화된 추출             │
    │  │ - 요약 생성                │
    │  └──────────────────────────┘
    │  ┌──────────────────────────┐
    │  │ 소형 모델 / 증류 모델        │
    │  │ - 텍스트 분류               │
    │  │ - 키워드 매칭               │
    │  │ - 형식 검증                │
    │  └──────────────────────────┘
    └─────────────────────────────────→ 호출 빈도

증류: 작은 모델이 전문 능력을 갖추게 하기

증류의 핵심 아이디어: 대형 모델의 판단 능력을 소형 모델에 “복사”하는 것입니다.

교사 모델 (GPT-4)           학생 모델 (0.6B 파라미터)
      │                         │
      │  레이블 데이터 생성          │
      ├─────────────────────────→│
      │  "요청 의도를 이해하라"      │  교사의 행동 패턴 학습
      │  [입력→출력 쌍]             │
      │                         │
      │  효과: 소형 모델이 특정 작업에서│
      │  교사 모델 수준에 근접        │

증류 vs 파인튜닝:

증류의 세 가지 경로:

추론 최적화

성능 최적화 프레임워크

LLM 추론 최적화를 네 가지 방향으로 나눕니다:

요청을 더 빠르게 처리하기

• 모델 소형화: 파라미터 수가 더 작은 모델 변형 선택
• 양자화: INT4/INT8/FP16 양자화로 계산 리소스 요구 감소
• 프루닝: 중복 가중치 제거, 모델 복잡도 감소
• 지식 증류: 대형 모델 데이터로 소형 모델 학습

양자화 정밀도 비교:
  FP32 (전체 정밀도)   → 기준 성능, 최대 계산 오버헤드
  FP16 (반정밀도)     → ~2x 가속, 정밀도 거의 손실 없음
  INT8               → ~4x 가속, 소폭 정밀도 손실
  INT4               → ~8x 가속, 정밀도 영향 주의 깊게 평가 필요

처리 요청 수 줄이기

• 컨텍스트 캐시(Context Cache): 멀티턴 대화의 공통 프리픽스 캐싱, 중복 계산 감소
• 배치 처리(Batching): 여러 요청을 하나의 배치로 병합, 하드웨어 활용률 향상
• 결과 캐시: 빈도 높은 동일 질문 결과 직접 캐시 반환

# 컨텍스트 캐시의 전형적 적용
# 멀티턴 대화에서 System Prompt + 과거 지식 문서가 공통 프리픽스
# 첫 번째 턴: 전체 계산 (전체 가격)
# 후속 턴: 캐시 히트 부분은 20% 가격으로 과금

Token 입출력 줄이기

• 입력 측: 입력 간소화, 중복 정보 제거, 긴 문서는 먼저 요약 생성
• 출력 측: Prompt를 통해 간결한 답변 유도, 합리적인 max_tokens 설정

max_tokens의 설계 철학: 이는 콘텐츠 제어 수단이 아닌 안전 밸브입니다. 의미적으로 완전한 짧은 응답은 Prompt 유도를 사용해야 하며, max_tokens는 비용 통제의 최후 방어선으로 더 적합합니다.

병렬화 처리

LLM 추론은 본질적으로 대규모 행렬 연산입니다. CPU와 GPU의 차이를 이해하십시오:

GPU 병렬화 전략:

• 데이터 병렬: 데이터 샤딩을 여러 GPU에 할당
• 모델 병렬: 모델의 다른 레이어를 다른 장치에 분산
• 파이프라인 병렬: 계산 과정을 단계로 나누어 순차 실행

대형 모델에 기본 의존하지 말 것

많은 시나리오에서 더 간단한 방법이 오히려 더 효율적입니다:

보안 가드레일

LLM이 직면한 보안 위협

LLM 애플리케이션은 다층적 보안 위협에 직면하며, 체계적인 방어 전략이 필요합니다:

방어 전략 매트릭스

보안 가드레일의 엔지니어링 구현

class SafetyGuard:
    """다층 보안 가드레일 구현"""

    def __init__(self):
        self.blocked_keywords = set()    # 사용자 정의 민감어
        self.rate_limits = {}            # 속도 제한 기록
        self.max_tool_calls = 10         # Agent 최대 도구 호출 횟수
        self.dangerous_commands = {      # 위험 명령 블랙리스트
            "rm -rf", "DROP TABLE", "DELETE FROM",
            "os.system", "subprocess.call", "eval("
        }

    def check_input(self, user_input: str) -> tuple[bool, str]:
        """입력 보안 검사"""
        # 1. 민감어 탐지
        for keyword in self.blocked_keywords:
            if keyword in user_input.lower():
                return False, f"입력에 민감어 포함: {keyword}"

        # 2. 명령 인젝션 탐지
        for dangerous in self.dangerous_commands:
            if dangerous.lower() in user_input.lower():
                return False, f"잠재적 위험 명령 탐지: {dangerous}"

        return True, "통과"

    def check_tool_call(self, tool_name: str, args: dict) -> tuple[bool, str]:
        """도구 호출 전 감사"""
        # 1. 호출 빈도 검사
        if self.rate_limits.get(tool_name, 0) >= self.max_tool_calls:
            return False, f"도구 {tool_name} 호출 횟수 초과"

        # 2. 파라미터 안전성 검사
        args_str = json.dumps(args).lower()
        for dangerous in self.dangerous_commands:
            if dangerous.lower() in args_str:
                return False, f"도구 파라미터에 위험 명령 포함: {dangerous}"

        self.rate_limits[tool_name] = self.rate_limits.get(tool_name, 0) + 1
        return True, "통과"

    def check_output(self, output: str) -> tuple[bool, str]:
        """출력 콘텐츠 심사"""
        # 출력해서는 안 되는 민감 정보 패턴 탐지
        sensitive_patterns = [
            r'\b\d{17}[\dXx]\b',           # 주민등록번호
            r'\b1[3-9]\d{9}\b',            # 휴대폰 번호
            r'[Pp]assword\s*[:=]\s*\S+',  # 비밀번호 패턴
        ]

        for pattern in sensitive_patterns:
            if re.search(pattern, output):
                return False, f"출력에 민감 정보 포함 가능: {pattern}"

        return True, "통과"

서킷 브레이커 메커니즘

Agent의 각 작업에 명시적 리소스 상한을 설정합니다:

class CircuitBreaker:
    """Agent 서킷 브레이커: 통제 불능 루프가 큰 손실을 초래하는 것을 방지"""

    def __init__(self,
                 max_api_calls: int = 10,      # 최대 API 호출 횟수
                 max_wall_time: int = 300,      # 최대 실행 시간 (초)
                 max_cost: float = 0.50):       # 최대 비용 (달러)
        self.max_api_calls = max_api_calls
        self.max_wall_time = max_wall_time
        self.max_cost = max_cost
        self.reset()

    def reset(self):
        self.api_calls = 0
        self.start_time = time.time()
        self.total_cost = 0.0

    def check(self) -> tuple[bool, str]:
        """서킷 브레이크 여부 확인"""
        self.api_calls += 1
        elapsed = time.time() - self.start_time

        if self.api_calls > self.max_api_calls:
            return False, f"API 호출 횟수 초과 ({self.api_calls}/{self.max_api_calls})"
        if elapsed > self.max_wall_time:
            return False, f"실행 시간 초과 ({elapsed:.0f}s/{self.max_wall_time}s)"
        if self.total_cost > self.max_cost:
            return False, f"비용 초과 (${self.total_cost:.2f}/${self.max_cost:.2f})"

        return True, "정상"

Harness Engineering

프로덕션 환경의 전체 청사진

Harness Engineering은 Agent 시스템이 개발에서 프로덕션까지 안정적으로 실행되도록 보장하는 풀셋 엔지니어링 실천입니다. 여기에는 다음이 포함됩니다:

가시성(Observability)

OpenTelemetry 표준을 사용하여 세 가지 유형의 데이터 수집을 구축합니다:

• Metrics(지표): Token 소비, 지연 시간 분포, 오류율
• Traces(추적): 한 번의 요청이 거치는 각 단계와 소요 시간
• Logs(로그): 각 단계의 입출력, 오류 스택, 감사 정보

# OpenTelemetry를 사용한 Agent 호출 계측
from opentelemetry import trace
from opentelemetry.instrumentation.openai import OpenAIInstrumentor

# OpenAI API 호출 자동 계측
OpenAIInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("agent_task")
def run_agent_task(task: str):
    # span이 자동으로 소요 시간과 컨텍스트 기록
    with tracer.start_as_current_span("llm_call") as span:
        span.set_attribute("task", task)
        result = agent.run(task)
        span.set_attribute("tokens_used", result.usage.total_tokens)
        return result

배포 전 체크리스트

Agent를 프로덕션에 배포하기 전에 다음 검사를 완료하십시오:

☐ SLO 정의
  ├─ TTFT (첫 Token 지연) 목표: _____ ms
  ├─ TPOT (Token당 생성 시간) 목표: _____ ms
  └─ 가용성 목표: _____%

☐ 비용 통제
  ├─ 단일 호출 예산 상한: $_____
  ├─ 일일 Token 소비 상한: _____ tokens
  └─ 경보 임계값 설정 완료

☐ 보안 방어
  ├─ 입력 보안 검사 활성화
  ├─ 출력 콘텐츠 심사 활성화
  ├─ Agent 행동 서킷 브레이커 설정 완료
  └─ 보안 모니터링 경보 설정 완료

☐ 재해복구 방안
  ├─ 모델 저하 경로 정의 완료
  ├─ 주요 경로 폴백 로직 검증 완료
  └─ 장애 복구 훈련 완료

☐ 평가 기준선
  ├─ 엔드 투 엔드 평가 점수: _____
  ├─ 구성 요소별 평가 점수: _____
  └─ 회귀 테스트 파이프라인 통합 완료

☐ 가시성
  ├─ OpenTelemetry 연동 완료
  ├─ 핵심 지표 Dashboard 구축 완료
  └─ 경보 규칙 설정 및 검증 완료

점진적 배포 전략

한 번에 전면 배포하지 마십시오. 점진적 전략을 채택하십시오:

Phase 1: 내부 테스트 (1-2주)
  └─ 팀원 사용, 초기 피드백 수집

Phase 2: 소규모 카나리 (5% 사용자)
  └─ 신구 시스템 평가 지표 비교

Phase 3: 범위 확대 (25% → 50% → 100%)
  └─ 각 단계에서 3-5일 머무르며 지표 관찰

Phase 4: 전면 배포
  └─ 모니터링 유지, 지속적 최적화 사이클 구축

프로덕션 운영 모범 사례

평가 기준선 관리:

• 현재 프로덕션 버전을 기준선으로 설정, 모든 새 버전은 기준선을 초과해야 함
• 정기적으로(주 단위) 최신 데이터로 기준선과 후보 버전 재테스트
• 배포 파이프라인에 기준선 검사 통합, 미달 버전 자동 차단

계층화된 저하 전략:

1. 주 모델 불가용 → 대체 모델로 전환
2. 대체 모델도 불가용 → 캐시된 일반 답변 사용
3. 캐시 미스 → 사전 설정된 저하 응답 템플릿 반환

비용 거버넌스:

• 모델별, 사용자별, 작업 유형별 차원에서 Token 소비 분석
• 이상 비용 급증 식별 및 경보 설정
• 정기 검토: 불필요한 긴 컨텍스트, 중복된 System Prompt가 있는가

모델 증류: 소형 모델에 도메인 전문 지식 가르치기

증류가 중요한 이유

대형 모델(GPT-4, Claude, Qwen-72B)은 뛰어난 품질을 제공하지만 추론 비용과 지연 시간이 높습니다. 분당 수천 건의 요청을 처리하는 프로덕션 시스템에서는 토큰 비용이 매우 커질 수 있습니다. 모델 증류는 실용적인 해결책을 제공합니다: 대형 모델의 출력을 훈련 데이터로 사용하여 소형 모델(7B-14B)이 훨씬 낮은 비용으로 동일한 동작을 재현하도록 가르칩니다.

교사 모델 (GPT-4, 175B 파라미터)
  │
  ├── 도메인 작업에 대한 고품질 응답 생성
  │
  ▼
훈련 데이터 (입력 → 교사 출력 쌍)
  │
  ├── 학생 모델 파인튜닝
  │
  ▼
학생 모델 (7B-14B 파라미터)
  │
  ├── 동일한 품질, 비용 10-50배 절감
  └── 지연 시간 5-10배 감소

증류 파이프라인

1단계: 작업 범위 정의

증류는 작업이 잘 정의되고 반복적일 때 가장 효과적입니다. 소형 모델이 수행해야 할 구체적인 도메인을 식별합니다:

# 예시: 고객 지원 의도 분류
task_definitions = [
    {
        "name": "intent_classification",
        "input_schema": {"user_message": "string", "context": "string"},
        "output_schema": {"intent": "string", "confidence": "float", "reasoning": "string"},
    },
    {
        "name": "response_generation",
        "input_schema": {"intent": "string", "knowledge": "string", "tone": "string"},
        "output_schema": {"response": "string", "sources": ["string"]},
    },
]

2단계: 교사 모델로 훈련 데이터 생성

대형 모델을 사용하여 도메인에 대한 고품질 예시를 생성합니다:

from openai import OpenAI

teacher = OpenAI(api_key="...")  # GPT-4 또는 유사 모델
student = OpenAI(base_url="http://localhost:8000/v1")  # 소형 모델

def generate_training_examples(task_def: dict, n_examples: int = 1000) -> list:
    examples = []
    for i in range(n_examples):
        # 다양한 입력 생성
        input_prompt = f"작업 '{task_def['name']}'에 대한 현실적인 입력을 생성하세요. 복잡도와 엣지 케이스를 다양하게 변화시키세요."
        input_resp = teacher.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": input_prompt}],
        )
        user_input = input_resp.choices[0].message.content

        # 교사 출력 생성
        teacher_resp = teacher.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"당신은 {task_def['name']}의 전문가입니다. 이 출력 스키마를 따르세요: {task_def['output_schema']}"},
                {"role": "user", "content": user_input},
            ],
        )
        teacher_output = teacher_resp.choices[0].message.content

        examples.append({"input": user_input, "output": teacher_output})

    return examples

training_data = generate_training_examples(task_definitions[0], n_examples=2000)

3단계: 학생 모델 파인튜닝

LoRA 또는 전체 파인튜닝을 사용하여 교사의 출력으로 소형 모델을 훈련합니다:

# Hugging Face Transformers + PEFT를 사용한 LoRA 파인튜닝
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model

model_name = "meta-llama/Llama-2-7b-hf"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# 효율적인 파인튜닝을 위해 LoRA 설정
lora_config = LoraConfig(
    r=16,  # LoRA 랭크
    lora_alpha=32,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 출력: trainable params: 4,194,304 || all params: 6,742,609,920 || trainable%: 0.0622

# 증류 데이터로 훈련
# ... (표준 훈련 루프)

4단계: 평가 및 반복

홀드아웃 테스트 세트에서 학생 모델의 출력을 교사의 출력과 비교합니다:

def evaluate_student_vs_teacher(test_set: list, teacher_client, student_client) -> dict:
    results = {"exact_match": 0, "semantic_similarity": 0, "total": len(test_set)}

    for example in test_set:
        student_resp = student_client.chat.completions.create(
            model="student-7b",
            messages=[{"role": "user", "content": example["input"]}],
        )
        student_output = student_resp.choices[0].message.content
        teacher_output = example["output"]

        # 정확 일치 검사
        if student_output.strip() == teacher_output.strip():
            results["exact_match"] += 1

        # 의미적 유사도 (embedding 사용)
        # ... (출력 간 코사인 유사도 계산)

    results["exact_match_rate"] = results["exact_match"] / results["total"]
    return results

# 목표: 교사와의 의미적 유사도 >85%

증류 should 할 때

프로덕션 모범 사례: 모니터링, 카나리 릴리스, A/B 테스트

가시성 스택

프로덕션 Agent 시스템에는 포괄적인 가시성이 필요합니다. 다음 모니터링 레이어를 구축합니다:

1. 애플리케이션 레벨 지표

import time
from prometheus_client import Counter, Histogram, Gauge

# 요청 지표
request_counter = Counter('agent_requests_total', '총 요청 수', ['model', 'intent', 'status'])
request_latency = Histogram('agent_request_duration_seconds', '요청 지연 시간', ['model', 'intent'])
active_sessions = Gauge('agent_active_sessions', '활성 세션 수')

# Token 사용량 지표
token_usage = Counter('agent_tokens_total', 'Token 사용량', ['model', 'type'])  # type: input/output

# 품질 지표
hallucination_rate = Gauge('agent_hallucination_rate', '환각 추정 비율')
user_satisfaction = Histogram('agent_user_satisfaction', '사용자 만족도 점수', buckets=[1, 2, 3, 4, 5])

def track_request(model: str, intent: str, status: str, latency: float, tokens_in: int, tokens_out: int):
    request_counter.labels(model=model, intent=intent, status=status).inc()
    request_latency.labels(model=model, intent=intent).observe(latency)
    token_usage.labels(model=model, type='input').inc(tokens_in)
    token_usage.labels(model=model, type='output').inc(tokens_out)

2. 분산 추적

OpenTelemetry를 사용하여 Agent 파이프라인 전체에서 요청을 추적합니다:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

def process_user_request(user_input: str):
    with tracer.start_as_current_span("process_request") as span:
        span.set_attribute("user.input_length", len(user_input))

        with tracer.start_as_current_span("retrieve_context") as retrieve_span:
            context = retrieve_relevant_docs(user_input)
            retrieve_span.set_attribute("docs.retrieved", len(context))

        with tracer.start_as_current_span("llm_generate") as llm_span:
            llm_span.set_attribute("model", "gpt-4")
            response = call_llm(user_input, context)
            llm_span.set_attribute("tokens.used", response.usage.total_tokens)

        return response

3. 로깅 전략

import structlog

logger = structlog.get_logger()

def log_agent_event(event_type: str, **kwargs):
    logger.info(
        event_type,
        session_id=kwargs.get("session_id"),
        user_id=kwargs.get("user_id"),
        model=kwargs.get("model"),
        intent=kwargs.get("intent"),
        latency_ms=kwargs.get("latency_ms"),
        tokens_in=kwargs.get("tokens_in"),
        tokens_out=kwargs.get("tokens_out"),
        error=kwargs.get("error"),
    )

# 사용 예시
log_agent_event(
    "request_completed",
    session_id="abc123",
    user_id="user_456",
    model="gpt-4",
    intent="faq_answer",
    latency_ms=1234,
    tokens_in=500,
    tokens_out=200,
)