Digital Nomad and FIRE

Time Travel

Time Travel 기능은 대화형 에이전트 개발 시 과거의 대화 상태로 되돌아가거나 대화 경로를 재탐색할 수 있는 도구로 에이전트의 의사 결정 과정을 분석하고, 다양한 시나리오를 테스트하며, 대화 흐름을 최적화할 수 있습니다.

Time Travel 기능의 주요 이점:

• 대화 흐름 분석: 과거의 대화 상태로 돌아가 에이전트의 반응과 의사 결정 과정을 상세히 검토할 수 있습니다.

• 시나리오 테스트: 다양한 대화 경로를 탐색하여 에이전트의 반응을 테스트하고 개선할 수 있습니다.

• 버그 수정 및 최적화: 특정 시점으로 되돌아가 문제를 재현하고 수정하여 에이전트의 성능을 향상시킬 수 있습니다.

Implementation

이전 코드는 질의가 발생하면, human_assistance 툴(Tool)에서 interrupt가 발생되도록 되어 있으므로, 해당 툴을 거치지 않도록 툴 등록을 해제 합니다.

# tools.py


# 도구 목록
tools = [TavilySearchResults(max_results=2)]

1. 그래프(Graph) 실행

get_state_history() 메서드를 사용하여 특정 스레드의 상태 이력을 가져올 수 있습니다. 스냅샷을 통해 원하는 시점으로 되돌아가거나 대화 경로를 재 탐색할 수 있습니다.

다음은 두개의 질의를 전달 한 후, 특정 영역의 state를 다시 실행할 수 있도록 지정합니다. (여기서는 6번째 tool 호출 결과 내역)

# agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_06"}}
    user_input = {"type": "user", "content": ("My Name is K. I'm learning LangGraph."
                  "Could you do some research on it for me?")}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()


    user_input = {"type": "user", "content": ("I would like to build chatbot agents using LangGraph.")}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()


    to_replay = None
    for state in graph.get_state_history(config):
        print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
        print("-" * 80)
        if len(state.values["messages"]) == 6:
            # We are somewhat arbitrarily selecting a specific state based on the number of chat messages in the state.
            to_replay = state

    print(to_replay.next)
    print(to_replay.config)

(결과)

# ~~ 생략 ~~

Num Messages:  8 Next:  ()
--------------------------------------------------------------------------------
Num Messages:  7 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  6 Next:  ('tools',)
--------------------------------------------------------------------------------
Num Messages:  5 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  4 Next:  ('__start__',)
--------------------------------------------------------------------------------
Num Messages:  4 Next:  ()
--------------------------------------------------------------------------------
Num Messages:  3 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  2 Next:  ('tools',)
--------------------------------------------------------------------------------
Num Messages:  1 Next:  ('chatbot',)
--------------------------------------------------------------------------------
Num Messages:  0 Next:  ('__start__',)
--------------------------------------------------------------------------------

{'configurable': {'thread_id': 'USER_06', 'checkpoint_ns': '', 'checkpoint_id': '1efe9bc1-fa9d-66aa-8006-85f90316985e'}}

메시지의 내용은 생략하고, 이후 결과들을 확인하면 6번째 호출은 도구를 호출 한 결과로써 to_replay.config 에는 checkpoint_id가 존재 하고 해당 id를 통해서 LangGraph의 체크포인터가 해당 시점의 state를 다시 가져올 수 있습니다.

checkpoint_id로 langsmith에서 확인할 수 있습니다.

Reference

• Time-Travel

Customizing State

LangGraph는 모델과 툴의 상태를 State로 정의하고, 이를 통해 상태의 변화를 추적합니다.
State의 변화 관리를 통해서 프로젝트의 요구 사항에 맞춰 상태를 효율적으로 관리하고 추적할 수 있게 해줍니다. 이를 통해 다양한 노드와 툴(Tool)이 상호작용하는 과정에서 발생할 수 있는 문제를 예방하고, 더 나은 사용자 경험을 제공할 수 있습니다.

Implementation

여기서는 이전 Agent의 내용을 바탕으로 챗봇이 검색 도구를 사용하여 특정 정보를 찾고 검토를 위해 사람에게 전달하는 시나리오를 구현합니다.

1. State에 키 추가

특정 정보를 찾아 State의 수정하기 위해서, 필요한 정보를 용이하게 접근할 수 있도록 state에 key 값을 추가합니다. 여기서는 name 과 birthday 키 값을 추가합니다.

# state.py


from typing import TypedDict, Annotated, Sequence
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    name: str
    birthday: str

2. State 업데이트

LangGraph에서 Human-in-the-loop 상호작용을 위한 human_assistance 도구를 다음과 같이 개선합니다.

✅ Call ID 기반의 State 업데이트

• Call ID를 활용하여 업데이트할 정확한 State를 찾도록 수정
• 각 실행 인스턴스를 구분하고 올바른 State 업데이트 보장

✅ 응답 정확도 판별 및 반환 값 변경

• y로 시작하는 응답("y")이면 결과를 반환, 아닌 경우에는 수정된 답변으로 반환
• 응답에서 "correct" 키가 없을 경우 기본값 빈 문자열 ("")을 반환

✅ InjectedToolCallId 사용하여 tool_call_id 관리

• LangChain 내부에서 tool_call_id를 자동 관리하도록 변경
• 각 툴의 실행 결과가 섞이지 않도록 분리하여 후속 작업의 혼선 방지

✅ Command를 활용한 State 업데이트

• State 변경 사항을 명확하게 추적 가능하도록 Command 사용

# tools.py

# 도구 정의
@tool
def human_assistance(name: str, birthday: str, tool_call_id: Annotated[str, InjectedToolCallId]) -> str:
    """Request assistance from a human."""
    human_response = interrupt({
        "question": "Is this correct?",
        "name": name,
        "birthday": birthday,
    })

    if human_response.get("correct", "").lower().startswith("y"):
        verified_name = name
        verified_birthday = birthday
        response = "Correct"
    else:
        verified_name = human_response.get("name", name)
        verified_birthday = human_response.get("birthday", birthday)
        response = f"Made a correction: {human_response}"

    state_update = {
        "name": verified_name,
        "birthday": verified_birthday,
        "messages": [ToolMessage(response, tool_call_id=tool_call_id)]  ,
    }

    return Command(update=state_update) # 상태 업데이트 명령어 반환

3. 그래프(Graph) 실행

상태(State) 업데이트 결과를 실행하여 확인합니다.
최근에 화두가 되었던, DeepSeek에 대한 출시일정을 확인하는 프롬프트(Prompt)를 구성해서 확인 해보겠습니다.

# agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_03"}}
    user_input = {"type": "user", "content": "Can you look up when DeepSeek was released? \
When you have the answer, use the human_assistance tool for review."}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()

    human_command = Command(
        resume={
            "name": "DeepSeek-R1",
            "birthday": "Jan 20, 2025"
        }
    )

    for event in graph.stream(human_command, config, stream_mode="values"):
        event["messages"][-1].pretty_print()

(결과)

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

Can you look up when DeepSeek was released? When you have the answer, use the human_assistance tool for review.
================================== Ai Message ==================================
Tool Calls:
  tavily_search_results_json (call_JK559rFs2TUTnOBXzq397glS)
 Call ID: call_JK559rFs2TUTnOBXzq397glS
  Args:
    query: DeepSeek release date
================================= Tool Message =================================
Name: tavily_search_results_json

[{"url": "https://en.wikipedia.org/wiki/DeepSeek", "content": "It would not be used for stock trading and would be
separate from High-Flyer's financial business.[4] In May 2023, the company was launched as DeepSeek.[2] DeepSeek's 
development is funded by High-Flyer.[3] After releasing DeepSeek-V2 in May 2024 which offered strong performance for 
a low price, DeepSeek became known as the catalyst for China's AI model price war. On 29 November 2023, DeepSeek 
launched DeepSeek LLM (large language model) which scaled up to 67B parameters. In December 2024, DeepSeek-V3 was 
launched. Benchmark tests showed it outperformed Llama 3.1 and Qwen 2.5 while matching GPT-4o and Claude 3.5 Sonnet.
[10][11] DeepSeek's optimization on limited resources highlighted potential limits of US sanctions on China's AI 
development.[12] \"Meet DeepSeek Chat, China's latest ChatGPT rival with a 67B model\". \"Chinese start-up DeepSeek's
new AI model outperforms Meta, OpenAI products\"."}, {"url": "https://www.techtarget.com/whatis/feature/DeepSeek-
explained-Everything-you-need-to-know", "content": "DeepSeek, a Chinese AI firm, is disrupting the industry with its 
low-cost, open source large language models, challenging U.S. tech giants. What is DeepSeek? DeepSeek focuses on 
developing open source LLMs. The company's first model was released in November 2023. DeepSeek Since the company was 
created in 2023, DeepSeek has released a series of generative AI models. The low-cost development threatens 
the business model of U.S. tech companies that have invested billions in AI. In contrast with OpenAI, which is 
proprietary technology, DeepSeek is open source and free, challenging the revenue model of U.S. companies charging 
monthly fees for AI services. Being based in China, DeepSeek challenges U.S. technological dominance in AI."}]
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_IBFdno2pWzb8gSFgFKfO7ySw)
 Call ID: call_IBFdno2pWzb8gSFgFKfO7ySw
  Args:
    name: DeepSeek
    birthday: November 2023
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_IBFdno2pWzb8gSFgFKfO7ySw)
 Call ID: call_IBFdno2pWzb8gSFgFKfO7ySw
  Args:
    name: DeepSeek
    birthday: November 2023
================================= Tool Message =================================
Name: human_assistance

Made a correction: {'name': 'DeepSeek-R1', 'birthday': 'Jan 20, 2025'}
================================== Ai Message ==================================

DeepSeek was launched in May 2023, with its first model released in November 2023. However, there was a correction 
indicating that the specific model "DeepSeek-R1" will be released on January 20, 2025.

결과를 확인하면, 검색 툴(tavily)을 통해 얻은 birthday는 November 2023 으로 human_assistance 툴에 interrupt 가 발생하였고, 리뷰를 통해 정정된 메시지로 전달 된 것을 확인 할 수 있다.
출시일을 birthday에 매핑되어 출력되어 의아하게 생각할 수 있는데, 이는 흔히 birthday 를 생일 뜻으로만 인식할 수 있는데 사전을 검색해보면 다른 뜻도 있음을 확인 할 수 있습니다.(부족한 영어실력이..발목을..😭😭)

3. 수동으로 State 업데이트 하기

graph.update_state를 활용하여 수동으로 thread의 키값으로 기반으로 state를 업데이트 할 수 있습니다.
기존 name 필드에 적용되어 있는 값을 DeepSeek-R1 Model로 이름을 변경하겠습니다.

# agent.py

# 그래프 실행
if __name__ == "__main__":

    print("================================= Graph State =================================")
    snapshot = graph.get_state(config)
    result = {k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
    print(result)

    print("============================ Manually Updating State ==========================")
    update_state = graph.update_state(config, {"name": "DeepSeek-R1 Model"})
    print(update_state)    
    snapshot = graph.get_state(config)

    result = {k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
    print(result)

(결과)

================================= Graph State =================================
{'name': 'DeepSeek-R1', 'birthday': 'Jan 20, 2025'}
============================ Manually Updating State ==========================
{'configurable': {'thread_id': 'USER_03', 'checkpoint_ns': '', 'checkpoint_id': '1efe8419-d950-6098-8006-3d30fc5fee6b'}}
{'name': 'DeepSeek-R1 Model', 'birthday': 'Jan 20, 2025'}

일반적으로 interrupt 기능을 사용하는 것이 권장 되며, 이를 통해 Human-in-the-loop 에서 필요한 정보만 전달하고, 불필요한 상태 변경을 방지할 수 있습니다.

4. LangGraph Studio에서 확인하기

LangGraph Studio에서 보다 편리하게 테스트 해볼 수 있습니다.

동일 질문의 질의하면, 검색 툴을 통해서 결과 반환된 후 Interrupt가 발생하고, Name과 Birthday를 입력할 수 있도록 커서가 이동 하였습니다.

업데이트 할 메시지를 입력하고, Resume을 클릭하며 내용의 변경사항과 최종 chatbot이 반환하는 메시지를 확인할 수 있습니다.

Reference

• Customizing State

AI 에이전트는 항상 예상대로 동작하지 않을 수 있으며, 일부 작업은 사람의 승인이 필요할 수도 있습니다. 이때, LangGraph의 Human-in-the-loop(HITL) 기능을 사용하면, 에이전트가 특정 작업을 완료하는 데 있어 사람의 입력을 받을 수 있게 되어 신뢰성과 제어를 향상시킬 수 있습니다.

Human-in-the-loop(HITL) 기능 개요

Human-in-the-loop(HITL)은 에이전트가 특정 작업을 수행하는 중간에 사용자의 승인을 받거나, 작업을 일시적으로 멈추고 사람의 피드백을 기다릴 수 있는 기능입니다. 이 기능을 통해 AI 시스템은 자동화된 작업 외에도 사람이 개입해야 하는 작업을 처리할 수 있으며, 결과적으로 더 높은 정확도와 안정성을 제공합니다.

LangGraph에서는 interrupt() 함수로 이러한 흐름을 제어할 수 있습니다. 이 함수는 작업 흐름을 일시적으로 멈추고, 사용자가 작업을 확인하거나 추가적인 입력을 제공할 수 있도록 합니다. 이후, 사용자가 입력을 제공하면, 작업이 다시 진행됩니다.

활용 시나리오

Human-in-the-loop(HITL) 기능은 여러 상황에서 유용하게 활용될 수 있으며, 이를 통해 자동화의 신뢰성을 높이고, 중요한 결정을 사람이 개입하여 더 정확하게 처리할 수 있습니다.

• 서비스 승인 및 배포: 자동화된 시스템이 서비스를 배포하기 전에, 관리자에게 승인을 요청하여 실수로 잘못된 배포가 이루어지지 않도록 합니다.

• 위험 관리: 중요한 결정을 내리기 전, 시스템이 자동으로 결정을 내리는 대신 사람의 의견을 반영하여 리스크를 최소화합니다.

• 고객 서비스: AI 챗봇이 고객과의 대화 중 민감한 정보를 처리할 때, 사람이 개입하여 확인 후 진행하는 방식으로 사용될 수 있습니다.

Implementation

LangGraph에서 Human-in-the-loop 기능을 활용하려면, 기존의 작업 흐름에서 특정 부분에 interrupt() 를 추가하면 됩니다. 이를 통해 에이전트가 중요한 결정을 내리기 전에 사람의 입력을 받을 수 있게 됩니다.

1. Human Assistant Tool 만들기

챗봇에서 사람의 개입을 할 수 있는 Tool을 정의하고, 도구목록에 human_assitance 함수를 추가합니다.

# tools.py

from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv
from langchain_core.tools import tool
from langgraph.types import interrupt

load_dotenv()

# 도구 정의
@tool
def human_assistance(query: str) -> str:
    """Request assistance from a human."""
    human_response = interrupt({"query": query})

    return human_response["data"]


# 도구 목록
tools = [TavilySearchResults(max_results=2), human_assistance]

2. 그래프 실행

#agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_02"}}
    user_input = {"type": "user", "content": "AI 에이전트를 개발하기 위해 전문적인 가이드를 받고 싶다."}

    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()


    print("================================= Graph State =================================")
    snapshot = graph.get_state(config)
    print(snapshot.next)

    # "AI 에이전트를 개발하려면 안정적이고 확장이 용이한 LangGraph를 확인해보시기를 권장합니다."
    human_response = (        
        "If you want to develop AI agents, we recommend checking out LangGraph, which is reliable and scalable."
    )

    human_command = Command(resume={"data": human_response})

    # human_command = None

    events = graph.stream(human_command, config, stream_mode="values")
    for event in events:
        if "messages" in event:
            event["messages"][-1].pretty_print()

(결과)

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

AI 에이전트를 개발하기 위해 전문적인 가이드를 받고 싶다.
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_Xyzp8czUPulqvGa9WsWT0qKf)
 Call ID: call_Xyzp8czUPulqvGa9WsWT0qKf
  Args:
    query: AI 에이전트 개발에 대한 전문적인 가이드 요청
================================= Graph State =================================
('tools',)
================================== Ai Message ==================================
Tool Calls:
  human_assistance (call_Xyzp8czUPulqvGa9WsWT0qKf)
 Call ID: call_Xyzp8czUPulqvGa9WsWT0qKf
  Args:
    query: AI 에이전트 개발에 대한 전문적인 가이드 요청
================================= Tool Message =================================
Name: human_assistance

If you want to develop AI agents, we recommend checking out LangGraph, which is reliable and scalable.
================================== Ai Message ==================================

AI 에이전트를 개발하기 위해 LangGraph를 추천합니다. 이 플랫폼은 신뢰할 수 있고 확장성이 뛰어나며, AI 에이전트를 구축하는 데 필요한 다양한 도구와 리소스를 제공합니다. LangGraph에 대한 자세한 정보와 사용 방법을 살펴보시면 좋겠습니다. 추가적인 도움이 필요하시면 말씀해 주세요!

• 결과를 확인하면, 사용자 질의(HumanMessage)가 발생한 후 human_assistance Tool에서 interrupt가 발생하여 그래프의 상태가 Tools에 머무른 것을 확인할 수 있습니다.
• 이후 그래프의 재개(Resume)를 하기 위해 Command의 인자값으로 resume 으로 전달하여 발생된 interrupt를 통과하여 그래프가 재개되었음을 확인할 수 있습니다.
• 최종적으로, 챗봇은 human_assistance의 Messsage를 통해서 최종적으로 결과를 전달하는 구조가 됨을 확인 할 수 있습니다.

Python의 내장 input()함수와 비슷하게, interrupt도구 내부에서 호출하면 실행이 일시 중지됩니다.
진행 상황은 체크포인터 선택에 따라 지속됩니다 .

4. LangSmith에서 추적 해보기

① Tool 호출단계에서 human_assitance에서 interrupt 가 발생한 부분을 확인할 수 있습니다.
② 그래프의 재개를 위해 Resume의 인자로 Data 키에 message가 들어간 부분을 확인할 수 있습니다.

5. LangGraph Studio에서 확인해보기

LagnGraph Studio에서 테스트를 수행해보면, 사용자 질의가 전달되면 Interrupt가 발생하여 답변이 멈춘상태를 확인할 수 있습니다.

Continue를 클릭하면 그래프 재개를 위한 Resume 박스가 만들어지고, 관련 메시지를 넣을 수 있습니다.

결과를 확인하면, human_assistance의 Tool에서 전달 메세지를 토대로 결과가 전달 되었음을 알 수 있습니다.

Reference

• Human-in-the-loop
• LangGraph-Types-Command
• 번역 LangGraph-Tutorial

LangGraph에서는 채크포인터(CheckPointer) 기능을 통해 챗봇의 상태를 저장하고 추적할 수 있습니다. 이를 활용하면 대화의 맥락을 유지하거나 에러 복구, Human-in-the-Loop(HITL), 그리고 Time-Travel 등 다양한 시나리오에서 유용하게 활용할 수 있습니다.

앞서 만든 Agent에서는 대화의 맥락을 이어갈 수 있는 형태의코드가 아니었습니다.

# agent.py

# ~~ 중략 ~~

# 그래프 빌더 초기화
graph_builder = StateGraph(AgentState, config_schema=GraphConfig)

# 'chatbot' 노드 추가
graph_builder.add_node("chatbot", chatbot)
# 'tool_node' 노드 추가
graph_builder.add_node("tools", tool_node)

# 'chatbot'을 엔트리포인트로 설정
graph_builder.set_entry_point("chatbot")

graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", END: END},
)

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")

# 그래프 컴파일
graph = graph_builder.compile()

(결과)

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

Hi there! My name is K.
================================== Ai Message ==================================

Hi K! How can I assist you today?
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

I don't have the capability to remember personal information or previous interactions,
so I don't know your name. However, you can share it with me if you'd like!

Implementation

1. 체크포인터 만들기

MemorySaver를 사용하여 인메모리 체크포인터를 사용합니다.

# agent.py

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)

• MemorySaver 객체 를 생성하고, checkpointer를 사용하여 그래프를 컴파일 및 상태를 저장합니다.

튜토리얼에서는 in-memory checkpointer 를 사용하지만,
프로덕션 단계에서는 SqliteSaver 또는 PostgresSaver 를 사용하여 자체 DB에 연결할 수 있습니다.

2. ToolNode와 tools_condition을 사용하여 그래프 구축하기

병렬 처리, 조건부 흐름 제어, 그래프 기반 동적 흐름 관리 등 시스템의 효율성, 확장성, 유연성을 향상시키고 다양한 동작을 효율적으로 처리할 수 있도록 하기 위해 prebuilt 된 ToolNode 와 tools_condition 으로 변경합니다.

• ToolNode : 그래프 상태(메시지 목록 포함)를 입력으로 받고 도구 호출의 결과로 상태 업데이트를 출력하는 LangChain Runnable
• tools_condition : state의 가장 최근 message를 가져와 tool_calls가 존재하는지 여부에 따라 분기

BasicToolNode → ToolNode로 변경

# nodes.py

import os, json
from langchain_openai import ChatOpenAI
from my_ai_agent.utils.state import AgentState
from my_ai_agent.utils.tools import tools
from dotenv import load_dotenv
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import ToolNode, tools_condition

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key= os.getenv("OPENAI_API_KEY"),
    max_tokens=None,
    temperature=0.7,
)

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

system_prompt = "Chat with the AI assistant. You can ask questions about anything else."

# 'chatbot' 노드
def chatbot(state):
    """상태에서 메시지를 받아서 LLM을 호출하는 함수"""
    messages = state["messages"]
    messages = [{"role": "system", "content": system_prompt}] + messages
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}


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

여기서부터는 gpt3.5로 하면 정확성이 너무 떨어집니다. 그나마 비용이 저렴한 40-mini로 바꿔서 진행합니다.

route_table → tools_condition 함수로 변경

# agent.py

import os
from typing import Literal, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from my_ai_agent.utils.state import AgentState
from my_ai_agent.utils.nodes import chatbot, tool_node
from langgraph.prebuilt import tools_condition


# Define the config
class GraphConfig(TypedDict):
    model_name: Literal["anthropic", "openai"]


# 그래프 빌더 초기화
graph_builder = StateGraph(AgentState, config_schema=GraphConfig)

# 'chatbot' 노드 추가
graph_builder.add_node("chatbot", chatbot)
# 'tool_node' 노드 추가
graph_builder.add_node("tools", tool_node)


# 'chatbot'을 엔트리포인트로 설정
graph_builder.set_entry_point("chatbot")


# 'chatbot'에서 'tool_node'로 이동하는 조건 추가
# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
    # It defaults to the identity function, but if you
    # want to use a node named something else apart from "tools",
    # You can update the value of the dictionary to something else
    # e.g., "tools": "my_tools"
    {"tools": "tools", END: END},
)

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")


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

LangGraph Cloud나 LangGraph Studio를 사용하는 경우, 그래프를 컴파일할 때 자동으로 checkpointer을 포함한 지속성 및 상태 관리를 처리하기 때문에 선언할 필요는 없습니다.

3. 그래프 실행

이전 메시지를 기억해서 답변 하는지 확인합니다.

# agent.py

# 그래프 실행
if __name__ == "__main__":

    config = {"configurable": {"thread_id": "USER_01"}}
    user_input = {"type": "user", "content": "Hi there! My name is K."}
    for event in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        event["messages"][-1].pretty_print()

    user_input = {"type": "user", "content": "Remember my name?"}
    for chunk in graph.stream({"messages": [user_input]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

(결과)

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

Hi there! My name is K.
================================== Ai Message ==================================

Hello, K! How can I assist you today?
================================ Human Message =================================

Remember my name?
================================== Ai Message ==================================

Yes, I remember your name is K. How can I assist you further?

4. Langgraph studio에서 확인해보기

LangGraph Studio를 열면 자동으로 새 스레드 창이 열립니다. 기존 스레드가 열려 있는 경우 다음 단계에 따라 새 스레드를 만듭니다.
우측 상단의 + 버튼을 클릭하여 새로운 스레드를 생성할 수 있습니다.

Reference

• Adding Memory to the chatbot
• 그래프에 영속성 추가하는 방법
• Migrating from ConversationalChain
• Prebuilt components

LangGraph 에서 Tool을 활용하여 어떻게 활용할 수 있는지에 대해 설명합니다. ChatBot이 답할 수 없는 질문을 처리하기 위해 웹 검색이 가능한 도구와 통합하여 관련정보를 찾고 더 나은 답변을 제공할 수 있도록 연계합니다.

Tool 사용하기

LangGraph에서 Tool은 함수나 API로 데이터를 처리하는 역할을 합니다. 예를 들어, 검색 기능을 구현하고 싶다면, 웹 검색 API나 데이터베이스 쿼리, 혹은 파일 시스템에서 데이터를 검색하는 도구를 활용할 수 있습니다.

Tool 설정하기

LangGraph에서 Tool을 설정하는 방법은 간단합니다. 먼저, Tool을 정의하고, 그것을 노드에서 사용할 수 있도록 연결해야 합니다.

환경설정

AI Agent 검색 엔진기반의 Tavily 검색엔진 패키지를 설치합니다.

poetry add tavily-python langchain_community

TAVILY API 키 설정

.env 에 TAVILY_API_KEY를 설정합니다.

TAVILY_API_KEY= "<your-api-key>"

Implementation

1. Tool 정의

검색 기능을 위한 Tool을 정의합니다.

# tools.py

from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv

load_dotenv()

tools = [TavilySearchResults(max_results=1)]

간단하게 Tavily 검색을 해볼 수 있습니다.

tools[0].invoke("what is Langchain?")

2. Tool 등록

LLM 모델에 bind_tools()를 사용하여 도구를 바인딩 합니다.
결과는 ToolCall Object 에 저장되고, .tool_calls 를 확인하여 도구 호출 결과를 확인할 수 있습니다.

# node.py

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

system_prompt = "Chat with the AI assistant. You can ask questions about anything else."

# 'chatbot' 노드
def chatbot(state):
    """상태에서 메시지를 받아서 LLM을 호출하는 함수"""
    messages = state["messages"]
    messages = [{"role": "system", "content": system_prompt}] + messages
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

3. Tool 노드(Node) 정의

가장 최근의 메시지에 tool_calls가 포함되어 있는 경우, 도구를 호출하는 노드를 구현합니다.

# node.py

from langchain_core.messages import ToolMessage

# 도구 노드 정의
class BasicToolNode:

    def __init__(self, tools: list) -> None:
        self.tool_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No messages found in input")
        print(message)
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tool_by_name[tool_call["name"]].invoke(tool_call["args"])

            outputs.append(ToolMessage(
                content=json.dumps(tool_result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"]
            ))

        return {"messages": outputs}

# 도구 노드 생성    
tool_node = BasicToolNode(tools)

4. 그래프(Graph) 정의

Tool 노드를 그래프에 추가 합니다.

# agent.py

from my_ai_agent.utils.node import tool_node

graph_builder.add_node("tools", tool_node)

도구 노드가 추가되면 conditional_edges를 정의할 수 있습니다.

5. 조건부엣지(Conditional edges) 정의

조건부 엣지(Conditional edges) 는 현재의 그래프의 상태에 따라 다른 노드로 라우팅을 할 수 있습니다.

• 조건부 엣지는 단일 노드에서 시작합니다. 즉, chatbot 노드가 실행될 때 다음 동작을 어떻게 처리할지 결정하는 역할을 합니다.
• 다음에 호출할 노드를 나타내는 문자열이나 문자열 목록을 반환함.

① 챗봇의 마지막 상태에서 tool_calls 가 있으면, tools 호출하는 함수를 생성합니다.
함수안에 도구 호출이 이루어지지 않으면 END 를 반환 하므로 finish_point 를 명시적으로 설정할 필요가 없습니다.

# agent.py

from langgraph.graph import END

def route_tools(state: AgentState):
    """
    Use in the conditional_edge to route to the ToolNode if the last message has tool calls. Otherwise, route to the end.
    """
    if isinstance(state, list):
        ai_messages = state[-1]
    elif messages := state.get("messages", []):
        ai_messages = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to too_edge: {state}")

    if hasattr(ai_messages, "tool_calls") and len(ai_messages.tool_calls) > 0:  # Check if the last message has tool calls
        return "tools"

    return END

② 챗봇 노드가 완료될 때마다 이 함수(route_tools)를 통해 다음으로 어디로 가야 하는지 확인하도록 그래프에 선언합니다. (add_conditional_edges)

# agent.py

# 'chatbot'에서 'tool_node'로 이동하는 조건 추가
# The `tools_condition` function returns "tools" if the chatbot asks to use a tool, and "END" if
# it is fine directly responding. This conditional routing defines the main agent loop.
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    # The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
    # It defaults to the identity function, but if you
    # want to use a node named something else apart from "tools",
    # You can update the value of the dictionary to something else
    # e.g., "tools": "my_tools"
    {"tools": "tools", END: END},
)

③ 그리고, chatbot 노드에서 tool 이 호출되고 난 이후에 chatbot 노드가 다음 스텝을 결정할 수 있도록 경로를 설정합니다.

# agent.py

# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")

그래프 컴파일(Compile)

compile() 메서드는 정의된 그래프를 실행 가능한 형태로 변환합니다.

# agent.py

graph = graph_builder.compile()

그래프의 실행결과를 확인합니다. (LangStudio)

Reference

• Tavily Search
• Enhancing the Chatbot with Tools

LangGraph를 사용하여 프로젝트 구조부터 시작하여, 그래프의 정의, 노드 연결, 실행까지 AI 에이전트를 구축하는 과정을 단계별로 설명합니다.
그래프의 실행은 LangGraph Studio를 활용하여 그래프의 연결상태나 노드에 대한 구성들을 확인합니다.

프로젝트 구조

프로젝트 구조는 LangGraph Platform Application 구조를 참고하였습니다.

my-app/
├── my_agent # 모든 프로젝트 코드
│   ├── utils # 그래프를 위한 유틸리티
│   │   ├── __init__.py
│   │   ├── tools.py # 그래프를 위한 도구들
│   │   ├── nodes.py # 그래프를 위한 노드 함수들
│   │   └── state.py # 그래프의 상태 정의
│   ├── __init__.py
│   └── agent.py # 그래프 구성 코드
├── .env # 환경 변수
├── langgraph.json  # LangGraph 구성 파일
└── pyproject.toml # 의존성 관리 파일

• 📁 my_agent/ : LangGraph 프로젝트의 코드가 위치하는 핵심 디렉토리입니다. 비즈니스 로직과 그래프 구성 요소들이 포함됩니다.
• 📁 utils/ : 그래프를 구성하는 데 필요한 유틸리티 파일들(도구, 노드, 상태 정의)
• 📋 agent.py : LangGraph의 그래프를 구성하고 실행하는 코드가 들어 있습니다.
• 📋 langgraph.json : LangGraph의 그래프 구성과 관련된 설정 파일입니다.
• 📋 pyproject.toml : poetry기반의 프로젝트 의존성 및 설정을 관리합니다.

환경설정

프로젝트 시작을 위해 필요한 패키지를 설치하고, 개발환경을 구성합니다. 파이썬 패키지 관리는 Poetry로 관리합니다.
PC환경 세팅(MaC 기준)은 Poetry - 패키지 관리 도구 참고합니다.

프로젝트 폴더를 구성하고, pyproject.toml config를 설정합니다.

poetry init

다음 패키지 설치를 진행합니다.

poetry add langchain langchain-core langgraph langsmith langchain-openai python-dotenv

LangSmith 추적 설정하기

LangSmith는 LLM 기반 애플리케이션 라이프사이클의 모든 단계를 위한 개발자 플랫폼으로 디버그, 협업, 테스트 및 모니터링 기능을 제공합니다.

.env 파일에 LangSmith 에서 발급받은 키와 프로젝트 정보를 입력합니다.

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=<your-api-key>

OpenAI API 키 설정

.env 파일에 OPENAI_API_KEY 입력합니다.

OPENAI_API_KEY= "<your-api-key>"

가상환경 활성화

설치된 패키지의 실행환경을 활성화 합니다.

poetry env activate

Implementation

1. Initialize graph with state.

Graph에서 사용할 State를 정의 합니다.

# state.py

from typing import TypedDict, Annotated, Sequence
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

Agent 가 구동되어 실행될 때, 정의된 그래프로 LangGraph 애플리케이션에서 사용하도록 정의합니다.

# agent.py

import os
from typing import Literal, TypedDict
from langgraph.graph import StateGraph
from my_ai_agent.utils.state import AgentState

# Define the config
class GraphConfig(TypedDict):
    model_name: Literal["anthropic", "openai"]


# 그래프 빌더 초기화
graph_builder = StateGraph(AgentState, config_schema=GraphConfig)

2. Initialize the model.

Agent 에 사용할 LLM 모델을 선언합니다. 여기서는 ChatOpenAI를 사용합니다.

## node.py

import os
from langchain_openai import ChatOpenAI
from my_ai_agent.utils.state import AgentState
from dotenv import load_dotenv

load_dotenv()


llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    api_key= os.getenv("OPENAI_API_KEY"),
    max_tokens=None,
    temperature=0.7,
)

3. 노드(Nodes) 정의

작업을 수행할 노드를 정의합니다. 여기서는 chatbot이라는 역할을 수행하는 함수를 정의하였습니다.
노드는 2가지 역할을 수행합니다.

• 어떤 작업을 수행할지(수행할 경우) 결정하는 역할
• Agent가 액션을 취하기로 결정하였을 때 액션을 실행하는 역할

# node.py

system_prompt = "Chat with the AI assistant. You can ask questions about anything else."

def chatbot(state):
    """상태에서 메시지를 받아서 LLM을 호출하는 함수"""
    messages = state["messages"]
    messages = [{"role": "system", "content": system_prompt}] + messages
    response = llm.invoke(messages)
    return {"messages": [response]}

4. 진입점(Entry Point) 및 엣지(Edge) 정의

먼저, 그래프 실행 시 진입점인 agent노드를 설정해야 합니다.
엣지(Edge)는 LangGraph에서 노드 간의 연결을 나타냅니다.
조건부 엣지(Conditional Edge)는 목적지가 그래프 상태의 내용에 따라 다른 노드로 연결할 수 있습니다.

• 조건부 에지(Conditional Edge):
  • a. 에이전트가 조치를 취하라고 말한 경우 도구를 실행하거나
  • b. 에이전트가 도구 실행을 요청하지 않은 경우 완료(사용자에게 응답)

# agent.py

# 'chatbot'을 엔트리포인트로 설정
graph_builder.set_entry_point("chatbot")

# 'chatbot'을 종료포인트로 설정
graph_builder.set_finish_point("chatbot")

5. 그래프 컴파일(Compile)

compile() 메서드는 정의된 그래프를 실행 가능한 형태로 변환합니다.

# agent.py

graph = graph_builder.compile()

6. 그래프 실행 확인

LangGraph Studio를 실행하기 위해 필요한 Configuration을 정의합니다.

# langgraph.json

{
  "dockerfile_lines": [],
  "graphs": {
    "agent": "./my_ai_agent/agent.py:graph"   # agent 파일의 위치와 컴파일된 그래프 변수
  },
  "env": ".env",                              # configuration 파일 위치
  "python_version": "3.12",                   # 실행되어야할 파이썬 버전
  "dependencies": [                           # 패키지 종속성
    "."
  ]
}

※ LangGraph Studio는 https://studio.langchain.com/ 에서 설치 가능합니다.

실행할 프로젝트 폴더를 열고, 그래프를 확인합니다.

왼쪽 Frame에서 메세지를 입력하면, 오른 Frame에서 결과를 확인할 수 있으며, 그래프의 각 노드별 실행단계들을 확인할 수 있습니다.

Reference

• LangGraph Quick Start
• How to Set Up a LangGraph Application for Deployment

Poetry

Poetry는 Python 프로젝트의 의존성 관리 및 패키지 관리를 위한 보다 현대적이고 강력한 도구입니다. pyproject.toml 파일을 사용하여 설정과 의존성을 관리합니다.

주요기능

장점:

• 강력한 의존성 해결: Poetry는 pyproject.toml 을 사용하여 의존성 해결과 관리에서 매우 효율적이고, 복잡한 의존성 충돌을 잘 해결합니다.

  • 이는 PEP 518 표준에 따라 Python 패키지의 구성을 정의하는 방법으로, 향후 여러 도구와의 호환성을 고려한 설정 방식입니다.

• 통합된 패키지 관리: 패키지 배포와 버전 관리를 한 번에 할 수 있습니다. publish 명령어로 직접 PyPI에 패키지를 배포할 수 있어 편리합니다.

• 자동화된 패키지 빌드: 프로젝트 빌드 과정에서 필요한 모든 파일과 의존성을 자동으로 생성하여 배포할 수 있습니다.

단점:

• 학습 곡선: Pipenv보다는 좀 더 많은 기능을 제공하지만, 그만큼 학습 곡선이 존재합니다.

• 호환성 문제: 일부 오래된 패키지나 라이브러리와의 호환성 문제를 겪을 수 있습니다.

설치방법

System requirements

• Python 3.9 이상
• Linux, macOS, Windows 지원

Installation

MacOS

1️⃣ poetry가 설치 되어 있지 않은 경우, 설치 합니다.

brew install poetry

2️⃣ 프로젝트 생성 (pyproject.toml)

poetry는 패키지관리를 위한 pyproject.toml 생성 부터 시작합니다.

① poetry init

패키지 생성을 위한 기본 정보를 대화형 기반으로 pyproject.toml 파일을 만들 수 있습니다.

poetry init

Options

• --name: Name of the package.
• --description: Description of the package.
• --author: Author of the package.
• --python Compatible Python versions.
• --dependency: Package to require with a version constraint. Should be in format foo:1.0.0.
• --dev-dependency: Development requirements, see --dependency.

② poetry new

Python 프로젝트에 적합한 디렉토리 구조를 만들어서 생성합니다.

poetry new <프로젝트 이름> 

as follows:

my-package
├── pyproject.toml
├── README.md
├── my_package
│   └── __init__.py
└── tests
    └── __init__.py

3️⃣ 패키지 관리

① 패키지 추가

poetry add <패키지명>

개발 의존성으로 패키지를 추가하고자 한다면, --dev 플래그 사용

poetry add --dev <패키지명>  

② 패키지 삭제

poetry remove <패키지명>

③ 패키지 조회

poetry show

poetry show --tree # 의존성 트리

poetry show --latest # 최신 버전도 함께 표시

poetry show --outdated # 업데이트가 가능한 패키지의 목록 표시

poetry show --no-dev # 개발 의존성을 제외하고 표시

4️⃣ 가상환경 생성 및 활성화

poetry env activate

Poetry 2.0 업데이트에 따른 주요 변경사항

poetry export:

• Poetry 1.2 버전부터 poetry export 명령은 이제 기본 기능이 아니라 플러그인으로 제공됩니다.
• poetry export를 사용하려면 poetry-plugin-export 플러그인을 별도로 설치해야 합니다.

poetry shell:

• poetry shell 명령어도 더 이상 Poetry의 기본 명령어로 제공되지 않습니다.
• 새로운 명령어인 poetry env activate를 사용하여 가상환경 활성화를 해야합니다.
• poetry shell 명령어를 사용하고 싶다면, poetry-plugin-shell 플러그인을 설치하여 사용할 수 있습니다.

Reference

• Poetry
• pipx

LangGraph

LangGraph는 LLM을 활용한 멀티 에이전트 시스템을 구축을 위한 라이브러리 입니다. 상태 추적과 복잡한 워크플로우 관리에 강점을 가지고 있으며, 이를 통해 하나의 애플리케이션에서 여러 에이전트가 동시 또는 순차적으로 상호작용하며 작업을 수행할 수 있습니다.

LanghGraph의 특징

• Cycles(반복문) and Branching(조건문) : 상태에 따라 동작을 다르게 하거나, 반복적인 작업 수행 지원
• Persistence(지속성) : 각 단계 별 완료될 때마다 상태를 자동으로 저장
  • 에러 복구, 일시 중단 후 재개, 타임 트래블(time travel), Human-in-the-loop 기능 구현이 가능
    • 타임 트래블(time travel) : 과거의 상태로 돌아가거나 이전에 수행된 작업으로 되돌림
• Human-in-the-Loop : 에이전트가 계획한 다음 작업을 승인하거나 편집 기능
• Streaming Support : 각 노드에서 생성되는 대로 스트림 출력 제공.
  • 특히 토큰 스트리밍(token streaming) 을 지원하여, 텍스트 기반의 작업이나 결과가 한 번에 모두 나오지 않고 실시간으로 처리 가능
• Integration with LangChain : LangChain 및 LangSmith 와 통합 용이

LangGraph의 구조

flowchart TD
    %% 노드 정의
    _start_("_start_") --> agent("agent")
    agent -->|continue| action
    agent -->|end| _end_("_end_")
    action --> agent

    %% 스타일 설정
    style _start_ fill:#3E2C58,stroke:#A78BFA,stroke-width:2px,color:white
    style agent fill:#3E2C58,stroke:#A78BFA,stroke-width:2px,color:white
    style action fill:#1C274F,stroke:#93C5FD,stroke-width:2px,color:white
    style _end_ fill:#1C1B19,stroke:#FACC15,stroke-width:2px,color:white

    %% 선 색상 설정
    linkStyle 0 stroke:#A78BFA,stroke-width:2px
    linkStyle 1 stroke:#93C5FD,stroke-width:2px,stroke-dasharray:5 5
    linkStyle 2 stroke:#FACC15,stroke-width:2px,stroke-dasharray:5 5
    linkStyle 3 stroke:#93C5FD,stroke-width:2px

• 노드 (Nodes): 그래프의 기본 단위로, 각 노드는 특정한 작업을 수행하거나, 데이터를 변환하는 기능을 담당합니다. 예) 데이터 전처리, 정보 추출, 텍스트 분석 등

• 엣지 (Edges): 노드들 간의 연결을 나타내며, 데이터가 어떻게 흐르고 변환되는지를 시각적으로 나타냅니다.

• 그래프 (Graph): 여러 개의 노드와 엣지가 결합된 구조로, 복잡한 작업 흐름을 전체적으로 다룹니다.

LangGraph의 코드 구성

1. State(상태) 정의

• 에이전트 간에 어떤 정보를 주고 받을지 정의. (즉, 노드간의 상태를 전달함.)

2. Node 선언

• LLM을 통해 어떤식으로 정보를 처리 할지를 정의

3. Edge

• 노드들 간의 연결을 나타내며, 데이터가 어떻게 흐르고 변환되는지를 시각적으로 나타냄

LangChain vs LangGraph

1. 애플리케이션 구조 :

• LangChain : 주로 단일 에이전트를 중심으로 파이프라인을 구성하는 데 중점을 둡니다. 이는 각기 다른 NLP 작업을 순차적으로 처리하는 데 적합합니다.

• LangGraph : 멀티 액터 시스템을 구축하여 여러 에이전트가 협력하거나 경쟁하면서 복잡한 작업을 처리하는 데 적합합니다. 이는 더 큰 시스템이나 복잡한 비즈니스 로직을 관리할 때 유리합니다.

2. 상태 추적 :

• LangChain : 상태 추적을 명시적으로 제공하지 않지만, 각 파이프라인의 출력은 이전 단계의 결과를 활용할 수 있습니다.

• LangGraph : 상태 기반 애플리케이션을 구축할 수 있으며, 각 에이전트가 상태를 추적하고 이를 바탕으로 의사 결정을 내리며 작업을 진행할 수 있습니다.

3. 용도와 목적 :

• LangChain : 텍스트 변환, 데이터 처리, API 호출 등 다양한 단일 작업 흐름을 처리하는 데 유용합니다.

chain = prompt | llm | parser

chain.invoke({"query": query})

• LangGraph : 멀티 에이전트 협업과 복잡한 워크플로우의 구축을 목표로 하며, 에이전트 간의 상호작용을 통한 분산 처리가 필요한 경우에 적합합니다.

Reference

• Langchain과 Langgraph의 차이점
• 랭그래프 소개
• LangGraph
• Lanchain Introduction

https://dart.fss.or.kr/info/main.do?menu=410