RAG2026-5일32차시 · 고도화된 Modular RAG6 / 18← 이전목차다음 →MetaContents

32차시. 고도화된 Modular RAG

3. 실습 ② — 조립과 실행

출처: 원본 PDF 32차시_ModularRAG종합.pdf 6쪽 · 쪽 그림 6쪽

📝 graph.py 계속

Python
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,
    }
← 이전 절32차시 · 2. 실습 ① — 통합 그래프 완성다음 절 →32차시 · 4. 실습 ③ — 경로 커버리지 테스트