Conditional Edge

---

❓Conditional Edge 만들기

 

1️⃣ 기존 간단한 그래프

graph_builder.add_node('retrieve', retrieve)
graph_builder.add_node('generate', generate)

from langgraph.graph import START, END

graph_builder.add_edge(START, 'retrieve')
graph_builder.add_edge('retrieve', 'generate')
graph_builder.add_edge('generate', END)
graph = graph_builder.compile()

 

2️⃣ Conditional Edge가 있는 그래프

여기서 중요한 점은, 기존 노드와 차이점을 알아야한다.

 

• 기존 노드의 리턴은 State다.
• 기존 노드는 .add_node를 통해 등록된다.
• Conditional Edge에 사용되는 노드에 일반적으로 등록되지 않는다.
• Conditional Edge에 사용되는 노드는 State가 아닌 노드이름을 리턴한다.

from langchain import hub
from typing import Literal

def check_doc_relevance(state: AgentState) -> Literal['generate', 'rewrite']:
    """
    주어진 state를 기반으로 문서의 관련성을 판단합니다.

    Args:
        state (AgentState): 사용자의 질문과 문맥을 포함한 에이전트의 현재 state.

    Returns:
        Literal['generate', 'rewrite']: 문서가 관련성이 높으면 'generate', 그렇지 않으면 'rewrite'를 반환합니다.
    """
    query = state['query']  # state에서 사용자의 질문을 추출합니다.
    context = state['context']  # state에서 문맥을 추출합니다.

    # 문서 관련성 판단 체인을 구성합니다.
    doc_relevance_chain = doc_relevance_prompt | llm
    
    # 질문과 문맥을 사용하여 문서의 관련성을 판단합니다.
    response = doc_relevance_chain.invoke({'question': query, 'documents': context})

    # 관련성이 높으면 'generate'를 반환하고, 그렇지 않으면 'rewrite'를 반환합니다.
    if response['Score'] == 1:
        return 'generate'
    
    return 'rewrite'

graph_builder.add_node('retrieve', retrieve)
graph_builder.add_node('generate', generate)
graph_builder.add_node('rewrite', rewrite)

# check_doc_relevance 노드 등록 ❌


from langgraph.graph import START, END

graph_builder.add_edge(START, 'retrieve')
graph_builder.add_conditional_edges('retrieve', check_doc_relevance) # conditional_edge 등록
graph_builder.add_edge('rewrite', 'retrieve')
graph_builder.add_edge('generate', END)

 

3️⃣ Conditional Edge 추가 고급편

이렇게 처리하면, 그래프의 가독성이 훨씬 좋아진다.

graph_builder.add_conditional_edges(
    'retrieve',             # 시작점: 문서를 찾아온 후
    check_doc_relevance,    # 판단 함수: 이 문서가 질문과 관련 있는지 검사
    {
        'relevant': 'generate', # 관련 있다면 -> 답변 생성 단계로!
        'irrelevant': END       # 관련 없다면 -> 여기서 종료(END)
    }
)

 

Conidtional Edge 사용 시, 노드를 추가하고 싶다면, 다음과 같이 노드와
실제 동작할 노드(등록 안할 노드)를 구분한다.

 

• 여기서 check_heplfulness는 등록될 노드
• check_heplfulness_grader는 실제 동작되고, 등록도지 않을 노드
• check_heplfulness -> check_heplfulness_grader로 감

def check_helpfulness_grader(state: AgentState) -> str:
    """
    사용자의 질문에 기반하여 생성된 답변의 유용성을 평가합니다.

    Args:
        state (AgentState): 사용자의 질문과 생성된 답변을 포함한 에이전트의 현재 state.

    Returns:
        str: 답변이 유용하다고 판단되면 'helpful', 그렇지 않으면 'unhelpful'을 반환합니다.
    """
    # state에서 질문과 답변을 추출합니다
    query = state['query']
    answer = state['answer']

    # 답변의 유용성을 평가하기 위한 체인을 생성합니다
    helpfulness_chain = helpfulness_prompt | llm
    
    # 질문과 답변으로 체인을 호출합니다
    response = helpfulness_chain.invoke({'question': query, 'student_answer': answer})

    # 점수가 1이면 'helpful'을 반환하고, 그렇지 않으면 'unhelpful'을 반환합니다
    if response['Score'] == 1:
        return 'helpful'
    
    return 'unhelpful'

def check_helpfulness(state: AgentState) -> AgentState:
    """
    유용성을 확인하는 자리 표시자 함수입니다. 
    graph에서 conditional_edge를 연속으로 사용하지 않고 node를 추가해
    가독성을 높이기 위해 사용합니다

    Args:
        state (AgentState): 에이전트의 현재 state.

    Returns:
        AgentState: 변경되지 않은 state를 반환합니다.
    """
    # 이 함수는 현재 아무 작업도 수행하지 않으며 state를 그대로 반환합니다
    return state

graph_builder.add_conditional_edges(
    'check_helpfulness',
    check_helpfulness_grader,
    {
        'helpful': END,
        'unhelpful': 'rewrite'
    }
)

---

✏️ 프롬프트 만들기

from langchain_core.prompts import PromptTemplate
from langchain import hub

generate_prompt = hub.pull("rlm/rag-prompt") # 허브에서 가져온 프롬프트

dictionary = ['사람과 관련된 표현 -> 거주자']

rewrite_prompt = PromptTemplate.from_template(f"""
사용자의 질문을 보고, 우리의 사전을 참고해서 사용자의 질문을 변경해주세요 
사전: {dictionary}                                           
질문: {{query}}
""") # 직접 작성하는 프롬프트

---

출처