d01 - 12. Capstone

S11 · 캡스톤 — 모든 조각을 하나의 에이전트로

Agentic RAG 실전 (14시간) · Day 2 · 마무리

S7(라우팅·재작성) + S8(자기 교정) + S9(재순위)를 하나의 그래프로 합치고, Naive RAG와 비교 평가(S10)해 개선을 수치로 확인한다. S0 .venv/.env 로 실행.

목표

• 라우팅 → 재작성 → (재순위) 검색 → 문서 평가/재검색 → 생성 → 답변 평가/재생성을 한 에이전트로 통합한다.
• Naive RAG 베이스라인과 같은 질문 세트로 비교한다.
• 평가기로 개선 폭을 측정하고 회고한다.

 

1. 통합 설계 — 지금까지의 조각 모으기

 START → router ──direct──▶ direct_answer ─────────────▶ END
            │ search
            ▼
        rewrite ▶ retrieve(재순위) ▶ grade_docs ──부실&여유──▶ rewrite (루프)
                                          │ 충분/소진
                                          ▼
                                       generate ▶ grade_answer ──환각──▶ regenerate ▶ END
                                                          │ 충실
                                                          ▼
                                                         END

• S7: router(구조화 출력 분류) + rewrite(검색어 정리)
• S9: retrieve 에 재순위 적용
• S8: grade_docs(부실→재검색 루프, 상한) + grade_answer(환각→재생성)

표준 루프가 아니므로 저수준 StateGraph로 직접 조립합니다.

 

2. 설치 · 키 · 연결

S0 .venv 커널로 실행. .env 키 로드 후 인덱스 연결과 재순위기를 준비한다.

import os
from getpass import getpass

# S0 프로젝트 루트의 .env 에서 키 로드 (없으면 입력으로 폴백)
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass
for name in ["OPENAI_API_KEY", "PINECONE_API_KEY"]:
    if not os.environ.get(name):
        os.environ[name] = getpass(f"{name}: ")
print("키:", {k: bool(os.environ.get(k)) for k in ["OPENAI_API_KEY", "PINECONE_API_KEY"]})

키: {'OPENAI_API_KEY': True, 'PINECONE_API_KEY': True}

 

 

 

import os
from pinecone import Pinecone
from langchain_pinecone import PineconeVectorStore, PineconeEmbeddings
from langchain_openai import ChatOpenAI

embeddings = PineconeEmbeddings(model="multilingual-e5-large")
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
index = pc.Index("agentic-rag-lab")
# ★ 네임스페이스 미지정 시 빈 기본 네임스페이스를 조회한다. S3에서 색인한 chunk_500 을 메인 코퍼스로 사용.
NAMESPACE = "chunk_500"
vector_store = PineconeVectorStore(index=index, embedding=embeddings, namespace=NAMESPACE)
try:
    _stats = index.describe_index_stats()
    _ns = getattr(_stats, "namespaces", None) or (_stats.get("namespaces") if isinstance(_stats, dict) else {}) or {}
    _cnt = {k: getattr(v, "vector_count", v.get("vector_count") if isinstance(v, dict) else None) for k, v in _ns.items()}
    print("네임스페이스별 벡터 수:", _cnt)
    if not _cnt.get(NAMESPACE):
        print(f"⚠️ '{NAMESPACE}' 네임스페이스에 데이터가 없습니다. S3의 색인 셀을 먼저 실행하세요.")
except Exception as _e:
    print("인덱스 통계 확인 생략:", _e)
llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
print("연결 완료")

네임스페이스별 벡터 수: {'chunk_1000': 18, 'chunk_1500': 12, 'chunk_200': 89, 'chunk_500': 34, 'demo': 3}
연결 완료

 

 

 

# 재순위기 준비 (S9) — 멀티-문서 질문을 위해 후보를 넓게(12) 잡고 상위 5개를 남긴다.
# (Naive 는 k=3 한 번 검색만 하므로, '넓게 검색 → 재순위 top-5'가 통합의 coverage 이점이 된다.)
from langchain_pinecone import PineconeRerank
reranker = PineconeRerank(model="bge-reranker-v2-m3", top_n=5)   # 여러 주제를 담을 수 있게 top_n 넉넉히

def retrieve_reranked(query, fetch_k=12):
    wide = vector_store.similarity_search(query, k=fetch_k)       # 후보를 넓게
    if not wide:
        return []
    return reranker.compress_documents(documents=wide, query=query)  # 재순위 후 상위 top_n개

 

 

 

3. 통합 에이전트 — 노드 정의

S7·S8·S9의 노드를 한곳에 모은다(자세한 설명은 각 교시 참고).

from pydantic import BaseModel, Field
from typing import Literal, TypedDict
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.graph import StateGraph, START, END

# --- 구조화 출력 틀 (S7 라우팅 / S8 평가) ---
class Route(BaseModel):
    destination: Literal["search", "direct"] = Field(description="문서 검색 필요시 search, 인사/일반상식이면 direct")
class DocGrade(BaseModel):
    relevant: bool = Field(description="검색 문서가 질문에 관련 있으면 true")
class AnswerGrade(BaseModel):
    grounded: bool = Field(description="답변이 근거에만 기반하면 true")

router_llm  = llm.with_structured_output(Route)
doc_grader  = llm.with_structured_output(DocGrade)
ans_grader  = llm.with_structured_output(AnswerGrade)

# ★ 라우팅 기준을 구체적으로 준다(S7과 동일). "질문을 분류하세요" 한 줄로는 대부분 direct 로 몰려
#    통합 에이전트가 검색을 건너뛰고 일반 상식으로 답해 버린다.
ROUTER_SYSTEM = (
    "당신은 '누구나테크' 사내 문서 도우미의 라우터입니다. 사용자 질문을 두 경로로 분류하세요.\n"
    "- search: 회사 정책·규정·절차·기술·FAQ에 관한 질문. 예) 휴가/연차, 재택·근무시간, 배포·릴리스, "
    "장애 대응, 보안·계정, 경비, 온보딩, 성과평가, 개인정보, 고객지원, 코드리뷰 등 사내 문서로 답해야 하는 질문.\n"
    "- direct: 인사·감사·잡담이나 회사와 무관한 일반 상식처럼 사내 문서가 필요 없는 질문.\n"
    "회사 생활·업무 규정과 조금이라도 관련 있으면 search 로 분류하세요."
)

class State(TypedDict):
    question: str; route: str; search_query: str; context: str
    answer: str; retries: int; docs_relevant: bool; grounded: bool

MAX_RETRIES = 2

def router_node(s):
    d = router_llm.invoke([SystemMessage(content=ROUTER_SYSTEM), HumanMessage(content=s["question"])])
    return {"route": d.destination, "retries": 0, "search_query": s["question"]}

def direct_node(s):
    return {"answer": llm.invoke(s["question"]).content}

def rewrite_node(s):
    p = f"질문을 벡터 검색에 좋은 간결한 검색어로 바꾸세요(핵심 명사, 한 줄).\n질문: {s['question']}\n검색어:"
    return {"search_query": llm.invoke(p).content.strip(), "retries": s.get("retries", 0) + 1}

def retrieve_node(s):
    docs = retrieve_reranked(s["search_query"])               # S9 재순위 검색
    return {"context": "\n\n".join(f"[{d.metadata.get('title')}] {d.page_content}" for d in docs)}

def grade_docs_node(s):
    r = doc_grader.invoke([SystemMessage(content="문서가 질문에 관련 있는지 평가."),
                           HumanMessage(content=f"[질문]\n{s['question']}\n\n[문서]\n{s['context']}")])
    return {"docs_relevant": r.relevant}

def generate_node(s):
    # 관련 근거가 있으면 근거로 답하고 참고 문서 표기.
    # 관련 근거를 못 찾았으면(상한 도달) 없는 내용을 지어내지 말고 참고 문서도 표기하지 않는다.
    if s.get("docs_relevant"):
        p = f"[근거]만 사용해 한국어로 답하고 끝에 참고한 문서 제목을 표기하세요.\n\n[근거]\n{s['context']}\n\n[질문]\n{s['question']}"
    else:
        p = ("아래 [근거]에 질문에 답할 내용이 없으면 '제공된 사내 문서에서 해당 내용을 찾지 못했습니다'라고만 "
             "답하고 참고 문서 제목은 표기하지 마세요. 근거에 답이 있으면 그 내용만 사용해 답하고 참고 문서 제목을 표기하세요."
             f"\n\n[근거]\n{s['context']}\n\n[질문]\n{s['question']}")
    return {"answer": llm.invoke(p).content}

def grade_answer_node(s):
    r = ans_grader.invoke([SystemMessage(content="답변이 근거에만 기반하는지 평가."),
                           HumanMessage(content=f"[근거]\n{s['context']}\n\n[답변]\n{s['answer']}")])
    return {"grounded": r.grounded}

def regenerate_node(s):
    p = f"[근거]에 있는 내용으로만 답하고, 없으면 '문서에서 확인되지 않습니다'라고 답하세요.\n\n[근거]\n{s['context']}\n\n[질문]\n{s['question']}"
    return {"answer": llm.invoke(p).content}

 

 

 

4. 통합 그래프 조립

def after_docs(s):
    if s["docs_relevant"]:
        return "generate"
    return "rewrite" if s.get("retries", 0) < MAX_RETRIES else "generate"

def after_answer(s):
    return END if s["grounded"] else "regenerate"

g = StateGraph(State)
for name, fn in [("router", router_node), ("direct", direct_node), ("rewrite", rewrite_node),
                 ("retrieve", retrieve_node), ("grade_docs", grade_docs_node),
                 ("generate", generate_node), ("grade_answer", grade_answer_node),
                 ("regenerate", regenerate_node)]:
    g.add_node(name, fn)

g.add_edge(START, "router")
g.add_conditional_edges("router", lambda s: s["route"], {"search": "rewrite", "direct": "direct"})
g.add_edge("direct", END)
g.add_edge("rewrite", "retrieve")
g.add_edge("retrieve", "grade_docs")
g.add_conditional_edges("grade_docs", after_docs, {"generate": "generate", "rewrite": "rewrite"})
g.add_edge("generate", "grade_answer")
g.add_conditional_edges("grade_answer", after_answer, {END: END, "regenerate": "regenerate"})
g.add_edge("regenerate", END)

final_agent = g.compile()
print("통합 에이전트 컴파일 완료")

통합 에이전트 컴파일 완료

 

 

 

from IPython.display import Image, display
try:
    display(Image(final_agent.get_graph().draw_mermaid_png()))
except Exception:
    print(final_agent.get_graph().draw_mermaid())

 

 

 

 

 

5. Naive RAG 베이스라인

가장 단순한 RAG(한 번 검색 → 바로 생성)를 비교 대상으로 둔다.

def naive_rag(question):
    docs = vector_store.similarity_search(question, k=3)         # 무조건 검색(재순위·평가 없음)
    context = "\n\n".join(f"[{d.metadata.get('title')}] {d.page_content}" for d in docs)
    p = f"[근거]를 참고해 답하세요.\n\n[근거]\n{context}\n\n[질문]\n{question}"
    return {"answer": llm.invoke(p).content, "context": context}

 

 

 

 

6. 비교 실행 — 멀티-문서 질문

단일 문서 질문은 Naive 도 잘 풀어 비교가 안 된다. 서로 다른 두 문서를 결합해야 답할 수 있는 질문과 정답 핵심 포인트(must) 라벨로 평가셋을 구성한다(라우팅 확인용 인사 1건 포함).

# ★ 멀티-문서 평가셋: '서로 다른 두 문서'를 결합해야 답할 수 있는 질문 + 정답 핵심 포인트(must) 라벨.
#   단일 문서 질문은 Naive(k=3)도 잘 풀어 비교가 안 되므로, 일부러 두 주제를 묶었다.
eval_set = [
    {"q": "재택근무 중 보안은 어떻게 지키고, 프로덕션 배포는 누가 최종 승인하나요?",
     "must": ["회사 지급 노트북·VPN 사용, 개인 기기로 사내 시스템 접속 금지",
              "엔지니어링 리드가 수동으로 최종 승인(2단계 게이트)"]},
    {"q": "신입 온보딩 첫 주에 무엇을 하고, 성과 평가는 얼마나 자주 이루어지나요?",
     "must": ["온보딩 버디 배정 및 개발 환경·코드베이스 학습",
              "성과 평가는 반기(상·하반기)마다 진행"]},
    {"q": "Sev1 장애의 1차 대응 시간과, 보안 사고가 의심될 때의 조치를 알려줘",
     "must": ["Sev1은 5분 이내 1차 대응 시작",
              "보안팀(security@)에 신고하고 해당 계정 세션 강제 종료"]},
]

# (참고) 라우팅 시연 — 인사는 검색 없이 direct 로 빠진다
g = final_agent.invoke({"question": "고마워요"})
print("[라우팅 확인] '고마워요' → 통합 경로:", g.get("route"), "/ 검색했나:", bool(g.get("context")), "\n")

for item in eval_set:
    q = item["q"]
    print("=" * 72)
    print("Q:", q)
    fin = final_agent.invoke({"question": q})
    nai = naive_rag(q)
    print("  통합 경로:", fin.get("route"))
    print("\n[통합 에이전트]", fin["answer"][:200], "...")
    print("\n[Naive RAG]   ", nai["answer"][:200], "...\n")

[라우팅 확인] '고마워요' → 통합 경로: direct / 검색했나: False 

========================================================================
Q: 재택근무 중 보안은 어떻게 지키고, 프로덕션 배포는 누가 최종 승인하나요?
  통합 경로: search

[통합 에이전트] 재택근무 중 보안은 회사가 지급한 노트북과 VPN을 사용해야 하며, 개인 기기에서 사내 시스템 접속은 금지된다. 카페 등 공용 공간에서는 화면 보안 필름을 사용하고 민감 정보 화면 공유를 피해야 한다. 화상회의 시 고객 정보나 내부 자료 공유 시 참석자 범위를 확인하고 녹화 여부를 사전에 고지해야 한다.

프로덕션 배포의 최종 승인은 릴리스 담당자가 배포  ...

[Naive RAG]    재택근무 중 보안은 다음과 같이 지킵니다:

- 회사가 지급한 노트북과 VPN을 사용해야 하며, 개인 기기로 사내 시스템에 접속하는 것은 금지됩니다.
- 카페 등 공용 공간에서 근무할 경우 화면 보안 필름을 사용하고, 민감 정보 화면 공유를 피해야 합니다.
- 화상회의에서 고객 정보나 내부 자료를 공유할 때는 참석자 범위를 확인하고 녹화 여부를 사전에 고지 ...

========================================================================
Q: 신입 온보딩 첫 주에 무엇을 하고, 성과 평가는 얼마나 자주 이루어지나요?
  통합 경로: search

[통합 에이전트] 신입 온보딩 첫 주에는 온보딩 버디가 배정되어 개발 환경 설정, 코드베이스 구조, 배포 절차를 함께 익히고, 매일 15분 체크인을 통해 막힌 부분을 해소합니다. 첫 주 금요일에는 팀 소개 세션과 제품 데모에 참석하며, 아키텍처 개요 문서를 읽고 궁금한 점을 정리해 질문합니다. 또한 사내 용어집과 자주 쓰는 문서 링크를 모은 시작 페이지가 제공됩니다.

성과 ...

[Naive RAG]    신입 온보딩 첫 주에는 다음과 같은 활동을 합니다.

- 온보딩 버디(같은 팀의 선임 개발자)가 배정되어 개발 환경 설정, 코드베이스 구조, 배포 절차를 함께 익힙니다.
- 매일 15분씩 버디와 체크인을 하며 막힌 부분을 빠르게 해결합니다.
- 첫 주 금요일에는 팀 소개 세션과 제품 데모에 참석하고, 아키텍처 개요 문서를 읽으며 궁금한 점을 정리해 질문합니 ...

========================================================================
Q: Sev1 장애의 1차 대응 시간과, 보안 사고가 의심될 때의 조치를 알려줘
  통합 경로: search

[통합 에이전트] Sev1 장애의 1차 대응 시간은 알림 수신 후 5분 이내이다. 보안 사고가 의심될 때는 즉시 보안팀(security@)에 신고하고 해당 계정의 세션을 강제 종료해야 한다. 신고는 처벌 대상이 아니며, 늦은 은폐보다 빠른 신고를 권장한다.  
[장애 대응 및 온콜 운영], [보안 및 계정 관리 규정] ...

[Naive RAG]    Sev1 장애의 1차 대응 시간은 알림 수신 후 5분 이내입니다.

보안 사고가 의심될 때는 즉시 보안팀(security@)에 신고하고, 해당 계정의 세션을 강제 종료해야 합니다. 신고는 처벌 대상이 아니며, 빠른 신고가 권장됩니다. ...

 

 

 

 

7. 품질 비교 — 완전성(completeness)

멀티-문서 질문에서는 필수 포인트를 빠짐없이 담았는가(완전성) 가 핵심이다. Naive(한 번 검색)는 한쪽 주제를 놓쳐 완전성이 낮아지기 쉽고, 통합(넓게 검색→재순위 top-5)은 양쪽을 담는다. faithfulness 는 둘 다 높을 수 있어 차이를 못 드러내므로, 정답 라벨 기반 완전성으로 비교한다.

from pydantic import BaseModel, Field

class Score(BaseModel):
    score: float = Field(ge=0.0, le=1.0, description="0~1 점수")
scorer = llm.with_structured_output(Score)

def faithfulness(question, answer, context):
    # 답변이 [근거]로 뒷받침되는 정도(근거에 없는 주장은 감점)
    r = scorer.invoke([
        SystemMessage(content="답변이 [근거]로 뒷받침되는 정도를 0~1로 채점(근거에 없는 주장은 감점)."),
        HumanMessage(content=f"[근거]\n{context}\n\n[질문]\n{question}\n\n[답변]\n{answer}"),
    ])
    return r.score

def completeness(answer, must_points):
    # 정답 '필수 포인트'가 답변에 포함된 비율(0~1). 멀티-문서 질문에서 한쪽 주제를 놓치면 낮아진다.
    joined = "\n".join(f"- {m}" for m in must_points)
    r = scorer.invoke([
        SystemMessage(content="아래 '필수 포인트'가 [답변]에 포함된 비율을 0~1로 채점하세요. 표현이 달라도 의미가 맞으면 포함으로 봅니다."),
        HumanMessage(content=f"[필수 포인트]\n{joined}\n\n[답변]\n{answer}"),
    ])
    return r.score

print("멀티-문서 질문 — 완전성(필수 포인트 포함률) · 충실성  (통합 vs Naive)")
print(f"{'질문':26s} | completeness 통합/Naive | faithful 통합/Naive")
print("-" * 80)
s_fin = s_nai = 0.0
for item in eval_set:
    q, must = item["q"], item["must"]
    fin = final_agent.invoke({"question": q})
    nai = naive_rag(q)
    c_fin = completeness(fin["answer"], must)
    c_nai = completeness(nai["answer"], must)
    f_fin = faithfulness(q, fin["answer"], fin.get("context", "")) if fin.get("context") else 1.0
    f_nai = faithfulness(q, nai["answer"], nai["context"])
    s_fin += c_fin; s_nai += c_nai
    print(f"{q[:24]:26s} | {c_fin:>5.2f} / {c_nai:>5.2f}         | {f_fin:>4.2f} / {f_nai:>4.2f}")
n = len(eval_set)
print("-" * 80)
print(f"{'평균 completeness':26s} | {s_fin/n:>5.2f} / {s_nai/n:>5.2f}")

print("\n해석")
print("- completeness(필수 포인트 포함률)가 핵심 지표다. 두 문서를 결합해야 하는 질문에서")
print("  Naive(한 번 검색 k=3)는 한쪽 주제만 담아 낮아지기 쉽고, 통합(넓게 검색→재순위 top-5)은 양쪽을 담아 높다.")
print("- faithfulness 는 둘 다 높을 수 있다(각자 가진 근거에는 충실). '빠뜨림'은 완전성이 드러낸다.")
print("- 즉 의미 있는 비교에는 ① 난이도 있는(멀티-문서) 질문 ② 정답 라벨(must) ③ 완전성 지표가 필요하다.")

멀티-문서 질문 — 완전성(필수 포인트 포함률) · 충실성  (통합 vs Naive)
질문                         | completeness 통합/Naive | faithful 통합/Naive
--------------------------------------------------------------------------------
재택근무 중 보안은 어떻게 지키고, 프로덕션   |  1.00 /  1.00         | 1.00 / 1.00
신입 온보딩 첫 주에 무엇을 하고, 성과 평   |  1.00 /  0.75         | 1.00 / 1.00
Sev1 장애의 1차 대응 시간과, 보안 사   |  1.00 /  1.00         | 1.00 / 1.00
--------------------------------------------------------------------------------
평균 completeness            |  1.00 /  0.92

해석
- completeness(필수 포인트 포함률)가 핵심 지표다. 두 문서를 결합해야 하는 질문에서
  Naive(한 번 검색 k=3)는 한쪽 주제만 담아 낮아지기 쉽고, 통합(넓게 검색→재순위 top-5)은 양쪽을 담아 높다.
- faithfulness 는 둘 다 높을 수 있다(각자 가진 근거에는 충실). '빠뜨림'은 완전성이 드러낸다.
- 즉 의미 있는 비교에는 ① 난이도 있는(멀티-문서) 질문 ② 정답 라벨(must) ③ 완전성 지표가 필요하다.

 

🔍 관찰 포인트

• 코퍼스에 없는 질문에서 통합 에이전트는 자기 교정으로 "문서에서 확인되지 않습니다" 류로 정직하게 답해 충실성이 유지되는 반면, Naive는 그럴듯한 오답을 낼 수 있다.
• 평가는 LLM 채점이라 매번 약간 다르다 — 여러 질문의 평균으로 본다.

 

 

 

 

8. 캡스톤 체크리스트 ✅

•  Pinecone 인덱스에 한국어 코퍼스 색인 (S2/S3)
•  라우팅 + 쿼리 재작성 (S7)
•  재순위 적용 검색 (S9)
•  문서 관련성 평가 → 재검색 루프 (S8/CRAG)
•  답변 충실성 평가 → 재생성 (S8/Self-RAG)
•  무한 루프 방지(재시도 상한)
•  Naive 대비 비교 평가 (S10)
•  (확장) 체크포인터로 멀티턴 / HITL / 스트리밍 (S4·S10)

 

 

 

9. 회고 — 더 생각해 볼 것

• 어디서 에이전트화가 효과적이었고, 어디서 과했나? (지연·비용 대비 품질)
• 평가 데이터셋을 더 잘 만들면(정답·근거 라벨) 측정 신뢰도가 올라갑니다.
• 운영 확장: DB 체크포인터, 관측(로깅·추적), 캐싱, 하이브리드 검색, 비용 모니터링.
• 안전: 민감 작업에 HITL, 프롬프트 인젝션 방어, 출처 표기.

 

 

 

과정 마무리

S0(환경) → S1(개념) → S2S3(검색·RAG 토대) → S4S5(LangGraph·첫 에이전트) → S6(표준화) → S7(라우팅·재작성) → S8(자기교정) → S9(재순위) → S10(프로덕션·평가) → S11(통합).

핵심 메시지: 단순하게 시작하고, 필요한 만큼만 똑똑하게. 검색을 도구로 두고, 판단·교정·측정을 한 겹씩 더해 견고한 한국어 Agentic RAG를 완성했습니다. 수고하셨습니다.