Self-RAG와 CRAG

def knowledge_refinement(documents, query):
    """문서를 분해하고 관련성 점수로 필터링"""
    refined_knowledge = []

    for doc in documents:
        # 1. 문서를 작은 단위로 분해
        strips = decompose_into_strips(doc)

        for strip in strips:
            # 2. 각 strip의 관련성 점수 계산
            score = calculate_relevance(strip, query)

            # 3. 임계값 이상인 strip만 유지
            if score > THRESHOLD:
                refined_knowledge.append(strip)

    # 4. 정제된 지식 조각들을 재조합
    return concatenate(refined_knowledge)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Literal

class CRAGState(TypedDict):
    query: str
    documents: list
    grade: Literal["correct", "incorrect", "ambiguous"]
    web_results: list
    final_response: str

def retrieve(state: CRAGState) -> CRAGState:
    """벡터 DB에서 문서 검색"""
    docs = vector_store.similarity_search(state["query"])
    return {**state, "documents": docs}

def grade_documents(state: CRAGState) -> CRAGState:
    """문서 관련성 평가"""
    grade = relevance_evaluator(state["query"], state["documents"])
    return {**state, "grade": grade}

def route_by_grade(state: CRAGState) -> str:
    """등급에 따라 다음 노드 결정"""
    if state["grade"] == "correct":
        return "refine"
    elif state["grade"] == "incorrect":
        return "web_search"
    else:  # ambiguous
        return "hybrid"

def web_search(state: CRAGState) -> CRAGState:
    """웹 검색 수행"""
    results = tavily_search(state["query"])
    return {**state, "web_results": results}

# 그래프 구성
workflow = StateGraph(CRAGState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("refine", knowledge_refinement)
workflow.add_node("web_search", web_search)
workflow.add_node("generate", generate_response)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", route_by_grade, {
    "refine": "refine",
    "web_search": "web_search",
    "hybrid": "refine"  # hybrid는 refine 후 web_search도 실행
})
workflow.add_edge("refine", "generate")
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()