패스트캠퍼스 환급챌린지 26일차 : 테디노트의 RAG 비법노트 강의 후기

본 포스팅은 패스트캠퍼스 환급 챌린지 참여를 위해 작성하였습니다

https://fastcampus.info/4n8ztzq

(~6/20) 50일의 기적 AI 환급반💫 | 패스트캠퍼스

 

패스트캠퍼스 환급챌린지 26일차!

 

1) 공부 시작 시간 인증

 

 

2) 공부 종료 시간 인증

 

 

3) 강의 수강 클립 인증

 

4) 학습 인증샷

 

 

5) 학습 통계

 

---

TIL (Today I Learned)

중간 단계의 상태(State) 수동 업데이트

• LangGraph는 중간 단계의 상태를 수동으로 업데이트 할 수 있는 방안을 제공하고 있습니다.
• 상태를 업데이트하면 에이전트의 행동을 수정하여 경로를 제어 할 수 있으며,
• 심지어 과거를 수정할 수도 있습니다.
• 이 기능은 에이전트의 실수를 수정 하거나, 대체 경로를 탐색 하거나,
• 특정 목표에 따라 에이전트의 동작을 변경 할 때 특히 유용합니다.

from typing import Annotated
from typing_extensions import TypedDict

from langchain_teddynote.tools.tavily import TavilySearch
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_teddynote.graphs import visualize_graph

########## 1. 상태 정의 ##########
# 상태 정의
class State(TypedDict):
    # 메시지 목록 주석 추가
    messages: Annotated[list, add_messages]

########## 2. 도구 정의 및 바인딩 ##########
# 도구 초기화
tool = TavilySearch(max_results=3)

# 도구 목록 정의
tools = [tool]

# LLM 초기화
llm = ChatOpenAI(model="gpt-4o-mini")

# 도구와 LLM 결합
llm_with_tools = llm.bind_tools(tools)

########## 3. 노드 추가 ##########
# 챗봇 함수 정의
def chatbot(state: State):
    # 메시지 호출 및 반환
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 상태 그래프 생성
graph_builder = StateGraph(State)

# 챗봇 노드 추가
graph_builder.add_node("chatbot", chatbot)

# 도구 노드 생성 및 추가
tool_node = ToolNode(tools=tools)

# 도구 노드 추가
graph_builder.add_node("tools", tool_node)

# 조건부 엣지
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

########## 4. 엣지 추가 ##########

# tools > chatbot
graph_builder.add_edge("tools", "chatbot")

# START > chatbot
graph_builder.add_edge(START, "chatbot")

# chatbot > END
graph_builder.add_edge("chatbot", END)

########## 5. 그래프 컴파일 ##########
# 메모리 저장소 초기화
memory = MemorySaver()

# 그래프 빌더 컴파일
graph = graph_builder.compile(checkpointer=memory)

########## 6. 그래프 시각화 ##########
# 그래프 시각화
visualize_graph(graph)

from langchain_core.runnables import RunnableConfig

# 질문
question = "LangGraph 가 무엇인지 조사하여 알려주세요!"

# 초기 입력 상태를 정의
input = State(messages=[("user", question)])

# config 설정
config = RunnableConfig(
    configurable={"thread_id": "1"},  # 스레드 ID 설정
)

# 그래프 채널 목록 출력
list(graph.channels)

['messages',
 '__start__',
 'chatbot',
 'tools',
 'start:chatbot',
 'branch:chatbot:tools_condition:chatbot',
 'branch:chatbot:tools_condition:tools']
 
 # 그래프 스트림 호출
events = graph.stream(
    input=input, config=config, interrupt_before=["tools"], stream_mode="values"
)

# 이벤트 반복 처리
for event in events:
    # 메시지가 이벤트에 포함된 경우
    if "messages" in event:
        # 마지막 메시지의 예쁜 출력
        event["messages"][-1].pretty_print()

================================ Human Message =================================

LangGraph 가 무엇인지 조사하여 알려주세요!
================================== Ai Message ==================================
Tool Calls:
  tavily_web_search (call_uAZwKKbpIcMsKKOHI6aFIMty)
 Call ID: call_uAZwKKbpIcMsKKOHI6aFIMty
  Args:
    query: LangGraph

• 현재 단계는 ToolNode 에 의해 중단되었습니다.
• 가장 최근 메시지를 확인하면 ToolNode 가 검색을 수행하기 전 query 를 포함하고 있음을 알 수 있습니다.
• 여기서는 query 가 단순하게 LangGraph 라는 단어만을 포함하고 있습니다. (기존의 질문은 "LangGraph 가 무엇인지 조사하여 알려주세요!" 였습니다.)
• 현재 단계는 ToolNode 에 의해 중단되었습니다.

# 그래프 상태 스냅샷 생성
snapshot = graph.get_state(config)

# 가장 최근 메시지 추출
last_message = snapshot.values["messages"][-1]

# 메시지 출력
last_message.pretty_print()

================================== Ai Message ==================================
Tool Calls:
  tavily_web_search (call_uAZwKKbpIcMsKKOHI6aFIMty)
 Call ID: call_uAZwKKbpIcMsKKOHI6aFIMty
  Args:
    query: LangGraph

StateGraph의 update_state 메서드

update_state 메서드는 주어진 값으로 그래프의 상태를 업데이트합니다. 이 메서드는 마치 as_node에서 값이 온 것처럼 동작합니다.

매개변수

• config (RunnableConfig): 실행 구성
• values (Optional[Union[dict[str, Any], Any]]): 업데이트할 값들
• as_node (Optional[str]): 값의 출처로 간주할 노드 이름. 기본값은 None

반환값

• RunnableConfig

주요 기능

1. 체크포인터를 통해 이전 상태를 로드하고 새로운 상태를 저장합니다.
2. 서브그래프에 대한 상태 업데이트를 처리합니다.
3. as_node가 지정되지 않은 경우, 마지막으로 상태를 업데이트한 노드를 찾습니다.
4. 지정된 노드의 writer들을 실행하여 상태를 업데이트합니다.
5. 업데이트된 상태를 체크포인트에 저장합니다.

주요 로직

1. 체크포인터를 확인하고, 없으면 ValueError를 발생시킵니다.
2. 서브그래프에 대한 업데이트인 경우, 해당 서브그래프의 update_state 메서드를 호출합니다.
3. 이전 체크포인트를 로드하고, 필요한 경우 as_node를 결정합니다.
4. 지정된 노드의 writer들을 사용하여 상태를 업데이트합니다.
5. 업데이트된 상태를 새로운 체크포인트로 저장합니다.

참고

• 이 메서드는 그래프의 상태를 수동으로 업데이트할 때 사용됩니다.
• 체크포인터를 사용하여 상태의 버전 관리와 지속성을 보장합니다.
• as_node를 지정하지 않으면 자동으로 결정되지만, 모호한 경우 오류가 발생할 수 있습니다.
• 상태 업데이트 중 SharedValues에 쓰기 작업은 허용되지 않습니다.

graph.update_state(
    # 업데이트할 상태 지정
    config,
    # 제공할 업데이트된 값. `State`의 메시지는 "추가 전용"으로 기존 상태에 추가됨
    {"messages": new_messages},
    as_node="tools",
)

print("(최근 1개의 메시지 출력)\\n")
print(graph.get_state(config).values["messages"][-1])

상태 업데이트는 그래프 단계를 시뮬레이션하므로, 해당하는 traces도 생성합니다.

messages를 사전 정의된 add_messages 함수로 Annotated 처리했습니다. (이는 그래프에 기존 목록을 직접 덮어쓰지 않고 항상 값을 추가합니다.)

동일한 논리가 여기에도 적용되어, update_state에 전달된 메시지가 동일한 방식으로 메시지가 추가하게 됩니다.

update_state 함수는 마치 그래프의 노드 중 하나인 것처럼 작동합니다! 기본적으로 업데이트 작업은 마지막으로 실행된 노드를 사용하지만, 아래에서 수동으로 지정할 수 있습니다. 업데이트를 추가하고 그래프에

사람에게 물어보는 노드

• 이런 상태 값들의 수정으로도 많은 것을 할 수 있지만, 메시지 목록에만 의존하지 않고 복잡한 동작을 정의하고자 한다면 상태에 추가 필드를 더할 수 있습니다.
• 이 튜토리얼에서는 새로운 노드를 추가하여 챗봇을 확장하는 방법을 설명합니다.
• 위의 예에서는 도구가 호출될 때마다 interrupt 를 통해 그래프가 항상 중단 되도록 Human-in-the-loop 을 구현하였습니다.
• 이번에는, 챗봇이 인간에 의존할지 선택할 수 있도록 하고 싶다고 가정해 봅시다.
• 이를 수행하는 한 가지 방법은 그래프가 항상 멈추는 "human" 노드 를 생성하는 것입니다.
• 이 노드는 LLM이 "human" 도구를 호출할 때만 실행됩니다.
• 편의를 위해, 그래프 상태에 "ask_human" 플래그를 포함시켜 LLM이 이 도구를 호출하면 플래그를 전환하도록 할 것입니다.

from typing import Annotated
from typing_extensions import TypedDict
from langchain_teddynote.tools.tavily import TavilySearch
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from pydantic import BaseModel
from langchain_openai import ChatOpenAI

class State(TypedDict):
    # 메시지 목록
    messages: Annotated[list, add_messages]
    # 사람에게 질문할지 여부를 묻는 상태 추가
    ask_human: bool
    
class HumanRequest(BaseModel):
    """Forward the conversation to an expert. Use when you can't assist directly or the user needs assistance that exceeds your authority.
    To use this function, pass the user's 'request' so that an expert can provide appropriate guidance.
    """
    request: str

# 도구 추가
tool = TavilySearch(max_results=3)

# 도구 목록 추가(HumanRequest 도구)
tools = [tool, HumanRequest]

# LLM 추가
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 도구 바인딩
llm_with_tools = llm.bind_tools(tools)

def chatbot(state: State):
    # LLM 도구 호출을 통한 응답 생성
    response = llm_with_tools.invoke(state["messages"])

    # 사람에게 질문할지 여부 초기화
    ask_human = False

    # 도구 호출이 있고 이름이 'HumanRequest' 인 경우
    if response.tool_calls and response.tool_calls[0]["name"] == HumanRequest.__name__:
        ask_human = True

    # 메시지와 ask_human 상태 반환
    return {"messages": [response], "ask_human": ask_human}
    
# 상태 그래프 초기화
graph_builder = StateGraph(State)

# 챗봇 노드 추가
graph_builder.add_node("chatbot", chatbot)

# 도구 노드 추가
graph_builder.add_node("tools", ToolNode(tools=[tool]))

Human 노드 설정

• 주로 그래프에서 인터럽트를 트리거하는 자리 표시자 역할을 합니다.
• 사용자가 interrupt 동안 수동으로 상태를 업데이트하지 않으면, LLM이 사용자가 요청을 받았지만 응답하지 않았음을 알 수 있도록 도구 메시지를 삽입합니다.
• ask_human 플래그를 해제하여 추가 요청이 없는 한 그래프가 노드를 다시 방문하지 않도록 합니다.

그래프에 인간 노드 추가

select_next_node는 플래그가 설정된 경우 human 노드로 경로를 지정합니다. 그렇지 않으면, 사전 구축된 tools_condition 함수가 다음 노드를 선택하도록 합니다.

tools_condition 함수는 단순히 chatbot이 응답 메시지에서 tool_calls을 사용했는지 확인합니다.

사용한 경우, action 노드로 경로를 지정합니다. 그렇지 않으면, 그래프를 종료합니다. from langgraph.graph import END

• chatbot 노드는 다음과 같은 동작을 합니다.
• 챗봇은 인간에게 도움을 요청할 수 있으며 (chatbot->select->human)
• 검색 엔진 도구를 호출하거나 (chatbot->select->action)
• 직접 응답할 수 있습니다 (chatbot->select->end).

일단 행동이나 요청이 이루어지면, 그래프는 chatbot 노드로 다시 전환되어 작업을 계속합니다.