d01 - 11. 평가

S10 · 프로덕션화 & 평가 — 영속성 · HITL · 스트리밍 · RAGAS

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

동작하는 에이전트를 운영 가능한 형태로 만든다: 영속성, 휴먼인더루프, 스트리밍, 평가. HITL·스트리밍 데모는 키 없이 돌고, 평가는 LLM을 쓴다. S0 .venv/.env 로 실행.

목표

• 프로토타입과 프로덕션의 차이를 이해한다.
• 휴먼인더루프(interrupt)와 스트리밍(stream)을 실행한다.
• 평가 지표를 이해하고, 현재 환경에서 도는 LLM 평가기로 품질을 측정한다(RAGAS는 표준 도구로 함께 소개).

 

1. 프로토타입 → 프로덕션

운영에 추가로 필요한 것:

• 영속성: 재시작·장애에도 대화·진행 상태 유지.
• HITL: 민감한 작업은 사람이 승인·수정.
• 스트리밍: 완성을 기다리지 않고 실시간 출력으로 체감 속도 개선.
• 평가: 개선 여부를 느낌이 아니라 숫자로 확인.

 

2. 영속성 — 체크포인터

S4의 체크포인터는 thread_id 별로 상태를 저장한다. InMemorySaver 는 메모리 저장이라 런타임 종료 시 소멸한다(실습용).

운영에서는 DB 기반 체크포인터를 쓴다.

• SQLite: langgraph-checkpoint-sqlite 의 SqliteSaver — 소규모·단일 서버.
• Postgres: langgraph-checkpoint-postgres 의 PostgresSaver — 다중 서버·대규모.

코드 구조는 동일하며 compile(checkpointer=...) 만 바꾼다. 이 교시 데모는 InMemorySaver.

 

3. 휴먼인더루프(HITL)

외부 전송·중요 변경·민감한 답 직전에 사람이 확인하거나 초안을 수정하게 한다.

노드 안에서 interrupt(...) 를 호출하면 그 지점에서 일시정지하고 사람에게 정보를 보여 준다. 사람 입력은 Command(resume=값) 으로 그 지점부터 이어 실행한다. 일시정지·재개에는 체크포인터가 필요하다.

아래는 "초안 생성 → 사람 승인/수정 → 확정" 의 최소 예제다(키 불필요).

 

4. 실행 준비

S0 .venv 커널로 실행. .env 키 로드(HITL·스트리밍은 키 불필요, 평가 단계에서 사용).

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}

 

HITL 실행 — 일시정지 후 사람 입력 대기

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command       # 일시정지(interrupt)와 재개(Command)

class ReviewState(TypedDict):
    draft: str       # 초안
    final: str       # 사람 검토 후 확정본

def make_draft(state):
    return {"draft": "연차는 입사 1년 차에 15일이 부여됩니다. (휴가 및 연차 정책)"}  # (데모용 고정 초안)

def human_review(state):
    # 여기서 그래프가 멈추고, 괄호 안 정보를 사람에게 보여 준다.
    decision = interrupt({"검토할 초안": state["draft"], "안내": "그대로면 'ok', 고치려면 수정문 입력"})
    # 사람이 'ok' 면 초안 그대로, 아니면 사람이 준 수정문을 확정본으로
    final = state["draft"] if decision.strip().lower() == "ok" else decision
    return {"final": final}

g = StateGraph(ReviewState)
g.add_node("make_draft", make_draft)
g.add_node("human_review", human_review)
g.add_edge(START, "make_draft"); g.add_edge("make_draft", "human_review"); g.add_edge("human_review", END)
app = g.compile(checkpointer=InMemorySaver())          # 일시정지/재개에는 체크포인터 필수

cfg = {"configurable": {"thread_id": "review-1"}}
paused = app.invoke({"draft": ""}, cfg)                # 1) 실행 → human_review에서 멈춤
print("일시정지됨. 사람에게 보여줄 내용:")
print(" ", paused["__interrupt__"][0].value)          # interrupt에 넣은 정보

일시정지됨. 사람에게 보여줄 내용:
  {'검토할 초안': '연차는 입사 1년 차에 15일이 부여됩니다. (휴가 및 연차 정책)', '안내': "그대로면 'ok', 고치려면 수정문 입력"}

 

# 2) 사람이 입력을 주고 재개 (여기선 수정문을 준다고 가정)
resumed = app.invoke(Command(resume="연차는 입사 1년 차 15일이며, 2년마다 1일씩 늘어 최대 25일입니다."), cfg)
print("확정본:", resumed["final"])

확정본: 연차는 입사 1년 차 15일이며, 2년마다 1일씩 늘어 최대 25일입니다.

 

🔍 관찰 포인트

• 첫 invoke 결과에 __interrupt__ 가 있다 → 사람을 기다리며 멈춤.
• Command(resume=...) 로 멈춘 지점(human_review)부터 이어서 끝까지 실행된다.

 

5. 스트리밍

답 완성까지 기다리면 멈춘 것처럼 느껴진다. 스트리밍은 중간 결과를 조각으로 흘려 체감 속도를 높인다.

graph.stream(...) 의 stream_mode:

• "updates": 노드별 진행(단계 표시).
• "values": 매 단계 전체 상태.
• "messages": LLM 토큰 실시간(타이핑 효과).

아래는 "updates" 데모(키 불필요). "messages" 모드는 LLM 토큰 스트리밍에 쓴다.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class Flow(TypedDict):
    step: int

def stage1(s): return {"step": 1}
def stage2(s): return {"step": 2}
def stage3(s): return {"step": 3}

gg = StateGraph(Flow)
gg.add_node("stage1", stage1); gg.add_node("stage2", stage2); gg.add_node("stage3", stage3)
gg.add_edge(START,"stage1"); gg.add_edge("stage1","stage2"); gg.add_edge("stage2","stage3"); gg.add_edge("stage3",END)
flow = gg.compile()

# stream(): 완료를 기다리지 않고 노드별 진행을 '조각'으로 받는다
for chunk in flow.stream({"step": 0}, stream_mode="updates"):
    print("진행 업데이트:", chunk)

진행 업데이트: {'stage1': {'step': 1}}
진행 업데이트: {'stage2': {'step': 2}}
진행 업데이트: {'stage3': {'step': 3}}

 

6. 평가 — 숫자로 확인

개선했으면 측정해야 한다. RAG 평가의 대표 지표(RAGAS가 표준):

지표	무엇을 보나
Faithfulness(충실성)	답이 근거에만 기반하는가
Answer Relevancy(적절성)	답이 질문에 직접 답하는가
Context Precision(정밀도)	검색 근거가 관련 있는가
Context Recall(재현율)	필요한 근거를 빠짐없이 가져왔는가(정답 근거 필요)

접근: RAGAS는 강력하지만 최신 LangChain 스택과 버전 충돌이 잦다(이 환경에서도 import 충돌 확인). 그래서 ① 현재 환경에서 도는 LLM 평가기를 직접 만들고, ② RAGAS 표준 코드는 전용/버전 고정 환경 실행용으로 참조 제시한다.

평가도 LLM 채점이라 완벽하지 않다. 절대 점수보다 개선 전/후 비교에 쓰는 것이 안전하다.

 

 

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_500': 34, 'chunk_200': 89, 'chunk_1000': 18, 'demo': 3, 'chunk_1500': 12}
연결 완료

 

LLM 평가기 — 구조화 출력 채점

세 지표를 0~1 점수로 받는다(S7·S8의 구조화 출력).

from pydantic import BaseModel, Field
from langchain_core.messages import SystemMessage, HumanMessage

class Score(BaseModel):
    """0~1 점수와 간단한 근거."""
    score: float = Field(ge=0.0, le=1.0, description="0(나쁨)~1(좋음) 사이 점수")
    reason: str = Field(description="채점 근거 한 문장")

scorer = llm.with_structured_output(Score)

def faithfulness(question, answer, contexts):
    ctx = "\n".join(contexts)
    r = scorer.invoke([
        SystemMessage(content="답변의 모든 주장이 [근거]로 뒷받침되는 정도를 0~1로 채점하세요. 근거에 없는 주장이 있으면 낮게."),
        HumanMessage(content=f"[근거]\n{ctx}\n\n[질문]\n{question}\n\n[답변]\n{answer}"),
    ])
    return r.score

def answer_relevancy(question, answer):
    r = scorer.invoke([
        SystemMessage(content="답변이 [질문]에 직접적으로 답하는 정도를 0~1로 채점하세요."),
        HumanMessage(content=f"[질문]\n{question}\n\n[답변]\n{answer}"),
    ])
    return r.score

def context_precision(question, contexts):
    # 각 근거가 관련 있는지 0/1로 보고 평균 → '검색 정밀도'
    vals = []
    for c in contexts:
        r = scorer.invoke([
            SystemMessage(content="이 [근거]가 [질문]에 답하는 데 관련 있으면 1, 무관하면 0에 가깝게 채점하세요."),
            HumanMessage(content=f"[질문]\n{question}\n\n[근거]\n{c}"),
        ])
        vals.append(r.score)
    return sum(vals) / len(vals) if vals else 0.0

 

 

평가 실행 — 좋은 답 vs 환각 답

같은 질문·근거에서 충실한 답과 지어낸 답의 점수 차이를 본다.

context = ["[휴가 및 연차 정책] 연차 휴가는 입사 1년 차에 15일이 부여되며, 근속 2년마다 1일씩 가산되어 최대 25일까지 늘어난다."]
question = "연차는 며칠인가요?"

good_answer = "입사 1년 차에 15일이며, 2년마다 1일씩 늘어 최대 25일입니다. (휴가 및 연차 정책)"
bad_answer  = "연차는 30일이며 매년 자동으로 두 배가 됩니다."   # 근거에 없는 환각

for label, ans in [("좋은 답", good_answer), ("환각 답", bad_answer)]:
    print(f"--- {label} ---")
    print(f"  faithfulness     : {faithfulness(question, ans, context):.2f}")
    print(f"  answer_relevancy : {answer_relevancy(question, ans):.2f}")
    print(f"  context_precision: {context_precision(question, context):.2f}\n")

--- 좋은 답 ---
  faithfulness     : 1.00
  answer_relevancy : 1.00
  context_precision: 1.00

--- 환각 답 ---
  faithfulness     : 0.00
  answer_relevancy : 0.80
  context_precision: 1.00

 

🔍 관찰 포인트

• 환각 답은 faithfulness 가 뚜렷이 낮다 → 자기 교정(S8) 효과의 수치화.
• 두 답 모두 질문엔 답하므로 answer_relevancy 는 비슷할 수 있다.
• context_precision 은 검색 품질(재순위 S9) 측정에 유용하다.

 

📚 참조 — RAGAS 표준 코드 (전용/버전 고정 환경에서 실행)

아래는 현재 RAGAS API입니다. 다만 최신 LangChain과 의존성 충돌이 잦으니, 별도 가상환경에서 버전을 고정해 실행하고 공식 문서로 최종 확인하세요.

# pip install ragas  (전용 환경 권장)
from ragas import evaluate, EvaluationDataset
from ragas.dataset_schema import SingleTurnSample
from ragas.metrics import Faithfulness, ResponseRelevancy, LLMContextPrecisionWithoutReference
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

evaluator_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4.1-mini"))
evaluator_emb = LangchainEmbeddingsWrapper(OpenAIEmbeddings(model="text-embedding-3-small"))

samples = [
    SingleTurnSample(
        user_input="연차는 며칠인가요?",
        response="입사 1년 차에 15일이며, 2년마다 1일씩 늘어 최대 25일입니다.",
        retrieved_contexts=["연차 휴가는 입사 1년 차에 15일이 부여되며, 2년마다 1일씩 가산..."],
        # reference="..."  # context_recall 등 일부 지표는 정답(reference) 필요
    ),
]
dataset = EvaluationDataset(samples=samples)
result = evaluate(
    dataset=dataset,
    metrics=[Faithfulness(), ResponseRelevancy(), LLMContextPrecisionWithoutReference()],
    llm=evaluator_llm,
    embeddings=evaluator_emb,
)
print(result)   # {'faithfulness': 0.89, 'answer_relevancy': 0.87, 'llm_context_precision_without_reference': 0.81}

우리가 만든 LLM 평가기는 RAGAS의 핵심 아이디어(LLM이 충실성·적절성·정밀도를 채점)를 축소판으로 구현한 것입니다. 개념을 이해하면 RAGAS도 쉽게 다룰 수 있습니다.

 

 

✏️ 미니 실습

1. S8 자기 교정 답과 Naive 답을 각각 채점해 faithfulness 차이를 확인한다.
2. S9 재순위 전/후 검색 결과로 context_precision 을 비교한다.
3. 스트리밍을 stream_mode="values" 로 바꿔 출력 차이를 본다.

# 모범 답안 1) S8 자기 교정 답과 Naive 답을 각각 채점해 faithfulness 차이를 확인한다

question = "연차는 며칠인가요?"
contexts = [
    "[휴가 및 연차 정책] 연차 휴가는 입사 1년 차에 15일이 부여되며, 근속 2년마다 1일씩 가산되어 최대 25일까지 늘어난다."
]

# 자기 교정된 답: 근거 안 정보만 사용
self_corrected_answer = (
    "연차는 입사 1년 차에 15일이 부여되며, 근속 2년마다 1일씩 늘어 최대 25일까지 가능합니다. "
    "(휴가 및 연차 정책)"
)

# Naive 답: 질문에는 답하지만 근거에 없는 내용을 섞음
naive_answer = (
    "연차는 기본 20일이고, 성과가 좋으면 추가 5일이 자동 지급됩니다."
)

for label, answer in [("S8 자기 교정 답", self_corrected_answer), ("Naive 답", naive_answer)]:
    print(f"--- {label} ---")
    print("  faithfulness     :", round(faithfulness(question, answer, contexts), 2))
    print("  answer_relevancy :", round(answer_relevancy(question, answer), 2))
    print("  context_precision:", round(context_precision(question, contexts), 2))
    print("  answer:", answer)
    print()

--- S8 자기 교정 답 ---
  faithfulness     : 1.0
  answer_relevancy : 1.0
  context_precision: 1.0
  answer: 연차는 입사 1년 차에 15일이 부여되며, 근속 2년마다 1일씩 늘어 최대 25일까지 가능합니다. (휴가 및 연차 정책)

--- Naive 답 ---
  faithfulness     : 0.0
  answer_relevancy : 1.0
  context_precision: 1.0
  answer: 연차는 기본 20일이고, 성과가 좋으면 추가 5일이 자동 지급됩니다.

실행 결과 해설

• 이 비교에서 가장 중요하게 봐야 할 값은 faithfulness다. 현재 출력처럼 자기 교정 답은 1.0, Naive 답은 0.0으로 갈리면, 두 답이 모두 질문에 답하는 척하더라도 근거를 따랐는지는 크게 다를 수 있다는 뜻이다.
• 반대로 answer_relevancy는 두 답 모두 높게 나올 수 있다. 즉, 질문에 겉으로는 잘 답해 보여도 사실과 근거를 어겼을 가능성은 별도로 봐야 한다. 그래서 운영 평가에서는 적절성만 보면 안 되고 충실성을 함께 봐야 한다.
• context_precision이 두 경우 모두 같게 나온 것도 자연스럽다. 이 지표는 답변 자체보다 '검색해 온 근거가 질문과 관련 있었는가'를 보는 값이기 때문이다. 즉, 이 셀은 S8의 자기 교정이 특히 faithfulness를 개선하는 데 의미가 있음을 보여준다.

 

 

 

# 모범 답안 2) S9 재순위 전/후 검색 결과로 context_precision 을 비교한다

from langchain_pinecone import PineconeRerank

eval_question = "재택근무할 때 지켜야 하는 보안 규칙은?"

# 재순위 전: 벡터 검색 상위 3개
before_docs = vector_store.similarity_search(eval_question, k=3)
before_contexts = [f"[{d.metadata.get('title', '제목 없음')}] {d.page_content}" for d in before_docs]

# 재순위 후: 상위 10개 후보를 모은 뒤 상위 3개로 재정렬
reranker = PineconeRerank(model="bge-reranker-v2-m3", top_n=3)
candidate_docs = vector_store.similarity_search(eval_question, k=10)
after_docs = reranker.compress_documents(candidate_docs, eval_question)
after_contexts = [f"[{d.metadata.get('title', '제목 없음')}] {d.page_content}" for d in after_docs]

before_score = context_precision(eval_question, before_contexts)
after_score = context_precision(eval_question, after_contexts)

print("Q:", eval_question)
print("재순위 전 context_precision:", round(before_score, 2))
for idx, text in enumerate(before_contexts, 1):
    print(f"  [전 {idx}]", text[:100].replace("\n", " "), "...")

print("\n재순위 후 context_precision:", round(after_score, 2))
for idx, text in enumerate(after_contexts, 1):
    print(f"  [후 {idx}]", text[:100].replace("\n", " "), "...")

Q: 재택근무할 때 지켜야 하는 보안 규칙은?
재순위 전 context_precision: 0.33
  [전 1] [재택근무 및 유연근무 제도] 재택 시에는 회사가 지급한 노트북과 VPN을 사용해야 하며, 개인 기기에서 사내 시스템에 접속하는 것은 금지된다. 카페 등 공용 공간에서 근무할 경우 ...
  [전 2] [재택근무 및 유연근무 제도] 재택근무는 주 최대 3일까지 가능하며, 매주 금요일은 전사 권장 재택일로 운영된다. 재택근무를 하려면 전주 목요일까지 다음 주 근무 형태를 그룹웨어에 ...
  [전 3] [재택근무 및 유연근무 제도] 월 1회 이상은 팀 동기화를 위해 사무실 출근이 권장되며, 분기마다 전사 오프사이트가 열린다. 팀별로 정기 대면 협업일을 지정해 운영할 수 있고, 신 ...

재순위 후 context_precision: 0.43
  [후 1] [재택근무 및 유연근무 제도] 재택 시에는 회사가 지급한 노트북과 VPN을 사용해야 하며, 개인 기기에서 사내 시스템에 접속하는 것은 금지된다. 카페 등 공용 공간에서 근무할 경우 ...
  [후 2] [재택근무 및 유연근무 제도] 재택근무는 주 최대 3일까지 가능하며, 매주 금요일은 전사 권장 재택일로 운영된다. 재택근무를 하려면 전주 목요일까지 다음 주 근무 형태를 그룹웨어에 ...
  [후 3] [보안 및 계정 관리 규정] 보안 사고가 의심되면 즉시 보안팀(security@)에 신고하고 해당 계정의 세션을 강제 종료한다. 신고는 처벌 대상이 아니며, 늦은 은폐보다 빠른 신 ...

 

 

실행 결과 해설

• 이 비교는 재순위가 답변을 직접 고치는 것이 아니라, 먼저 들어가는 근거의 질을 높인다는 점을 보여준다. 현재 실행에서도 context_precision이 0.33에서 0.43으로 올라갔으므로, 재순위 후 근거 집합이 질문에 더 밀접해졌다고 해석할 수 있다.
• 출력 내용을 보면 재순위 전에는 재택근무 일반 운영 규칙이 함께 섞여 있었고, 재순위 후에는 보안 및 계정 관리 규정처럼 질문의 보안 측면과 더 직접 관련된 문서가 상위권에 들어온다. 즉, 재순위는 같은 코퍼스 안에서도 더 답에 가까운 근거를 앞으로 당겨오는 역할을 한다.
• 이런 개선은 작아 보여도 운영에서는 중요하다. 검색 근거가 조금만 더 맞아도 뒤쪽 생성 단계의 환각 가능성이 줄고, S8 같은 자기 교정 루프가 불필요하게 도는 횟수도 줄일 수 있기 때문이다.

 

 

 

 

# 모범 답안 3) 스트리밍을 stream_mode="values" 로 바꿔 출력 차이를 본다

print("stream_mode='updates' 는 바뀐 조각만 보여줌")
for chunk in flow.stream({"step": 0}, stream_mode="updates"):
    print("  updates:", chunk)

print("\nstream_mode='values' 는 매 단계 전체 상태를 보여줌")
for chunk in flow.stream({"step": 0}, stream_mode="values"):
    print("  values:", chunk)

stream_mode='updates' 는 바뀐 조각만 보여줌
  updates: {'stage1': {'step': 1}}
  updates: {'stage2': {'step': 2}}
  updates: {'stage3': {'step': 3}}

stream_mode='values' 는 매 단계 전체 상태를 보여줌
  values: {'step': 0}
  values: {'step': 1}
  values: {'step': 2}
  values: {'step': 3}

실행 결과 해설

• stream_mode="updates"는 각 노드에서 바뀐 조각만 보여 준다. 그래서 현재 출력처럼 {'stage1': {'step': 1}} 같은 형태로 '무슨 단계가 끝났는지'를 빠르게 추적하기 좋다.
• 반면 stream_mode="values"는 매 단계의 전체 상태를 다시 보여 준다. 지금처럼 {'step': 0} → {'step': 1} → {'step': 2} → {'step': 3} 흐름이 그대로 보이므로, 디버깅할 때는 상태가 누적되며 어떻게 변하는지 이해하기 쉽다.
• 즉, updates는 진행 상황 모니터링에, values는 상태 디버깅에 더 적합하다. 운영 환경에서는 사용자에게는 messages나 간단한 진행 표시를 보여 주고, 개발 중 점검에는 values를 활용하는 식으로 목적을 나눠 쓰면 된다.

 

 

 

정리

• 영속성(체크포인터, 운영은 DB 기반), HITL(interrupt/Command), 스트리밍(stream_mode)으로 운영 요소를 갖췄다.
• 평가 지표를 이해하고 LLM 평가기로 좋은 답/환각 답을 수치로 구분했다. RAGAS는 표준 도구로 참조 코드를 제시했다.
• 다음(S11): 모든 조각을 하나의 에이전트로 통합하고 Naive 대비 개선을 측정한다.