32차시. 고도화된 Modular RAG
3. 실습 ② — 조립과 실행
출처: 원본 PDF 32차시_ModularRAG종합.pdf 6쪽 · 쪽 그림 6쪽
📝 graph.py 계속
def build_graph():
g = StateGraph(RAGState)
NODES = {
"classify": classifier_node,
"greeting": greeting_node,
"calc": calc_node,
"scope": scope_node,
"retrieve": retriever_node,
"generate": generator_node,
"verify": verifier_node,
"bump": bump_node,
"rewrite": rewrite_node,
"fallback": fallback_node,
}
for name, fn in NODES.items():
g.add_node(name, fn)
# ① 진입 - 유형 분류
g.add_edge(START, "classify")
g.add_conditional_edges("classify", route_by_intent, {
"greeting": "greeting", "calc": "calc",
"scope": "scope", "document": "retrieve",
})
for n in ("greeting", "calc", "scope"):
g.add_edge(n, END)
# ② 검색 후 분기
g.add_conditional_edges("retrieve", route_after_retrieve, {
OK: "generate", RESEARCH: "rewrite", GIVEUP: "fallback",
})
# ③ 생성 -> 검증
g.add_edge("generate", "verify")
# ④ 검증 후 분기
g.add_conditional_edges("verify", route_after_verify, {
DONE: END, REGEN: "bump",
RESEARCH: "rewrite", GIVEUP: "fallback",
})
# ⑤ 루프
g.add_edge("bump", "generate") # 재생성 루프
g.add_conditional_edges("rewrite", route_after_rewrite, {
"retry_search": "retrieve", # 재검색 루프
GIVEUP: "fallback",
})
g.add_edge("fallback", END)
return g.compile()
app = build_graph()
def ask(question: str, verbose: bool = False) -> dict:
t0 = time.time()
s = init_state(question)
s["deadline"] = t0 + MAX_TOTAL_SEC
final = app.invoke(s, {"recursion_limit": 40})
elapsed = time.time() - t0
if verbose:
print(f"\nQ: {question}")
for line in final.get("log", []):
print(f" · {line}")
return {
"answer": final.get("answer", ""),
"sources": [{"file": d.metadata.get("filename"),
"page": d.metadata.get("page_no")}
for d in final.get("documents", [])],
"intent": final.get("intent"),
"grade": final.get("grade"),
"verified": final.get("grade") == "pass",
"retries": final.get("retries", 0),
"rewrites": final.get("rewrites", 0),
"elapsed": round(elapsed, 2),
"log": final.get("log", []),
"_state": final,
}