[랭체인] LangGraph를 사용하여 고급 대화형 AI 에이전트 구축하기

안녕하세요, AI 개발자 여러분! 오늘은 LangGraph를 사용하여 지속성과 스트리밍 기능을 갖춘 고급 대화형 AI 에이전트를 구축하는 방법에 대해 알아보겠습니다. 이 기술을 통해 우리는 더 자연스럽고 효율적인 AI 상호작용을 만들 수 있습니다.

1. 소개

LangGraph는 복잡한 AI 워크플로우를 구축하기 위한 강력한 프레임워크입니다. 오늘 우리가 다룰 주요 기능은 다음과 같습니다:

• 지속성: 대화 상태를 유지하여 문맥을 기억하는 기능
• 스트리밍: 실시간으로 AI의 응답을 받는 기능

이러한 기능들은 더 자연스럽고 반응이 빠른 AI 에이전트를 만드는 데 큰 도움이 됩니다.

2. 환경 설정

먼저 필요한 라이브러리를 임포트합니다:

from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults

_ = load_dotenv()

여기서 우리는 LangGraph, LangChain, 그리고 기타 유용한 도구들을 사용하고 있습니다.

3. 에이전트 상태 정의

AI 에이전트의 상태를 정의합니다:

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

이 상태는 대화 기록을 저장합니다.

4. 지속성 설정

SQLite를 사용하여 대화 상태를 저장합니다:

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")

이렇게 하면 대화가 중단되더라도 상태를 유지할 수 있습니다.

5. AI 에이전트 클래스 구현

이제 AI 에이전트 클래스를 만들어 봅시다:

class Agent:
    def __init__(self, model, tools, checkpointer, system=""):
        self.system = system
        graph = StateGraph(AgentState)
        graph.add_node("llm", self.call_openai)
        graph.add_node("action", self.take_action)
        graph.add_conditional_edges("llm", self.exists_action, {True: "action", False: END})
        graph.add_edge("action", "llm")
        graph.set_entry_point("llm")
        self.graph = graph.compile(checkpointer=checkpointer)
        self.tools = {t.name: t for t in tools}
        self.model = model.bind_tools(tools)

    # ... (메서드 구현은 생략)

이 클래스는 대화 흐름을 관리하고, 필요할 때 외부 도구를 호출합니다.

6. 에이전트 초기화

에이전트를 초기화하고 사용할 준비를 합니다:

prompt = """당신은 똑똑한 연구 조수입니다. 검색 엔진을 사용하여 정보를 찾아보세요. \
여러 번 통화할 수 있습니다(함께 또는 순서대로). \
원하는 정보가 확실할 때만 정보를 조회합니다. \
후속 질문을 하기 전에 정보를 조회해야 하는 경우에는 조회할 수 있습니다!
"""
model = ChatOpenAI(model="gpt-4o")
abot = Agent(model, [tool], system=prompt, checkpointer=memory)

7. 대화 실행 및 지속성 테스트

이제 에이전트와 대화를 시작해 봅시다:

messages = [HumanMessage(content="What is the weather in sf?")]
thread = {"configurable": {"thread_id": "1"}}

for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v['messages'])

이렇게 하면 에이전트가 날씨 정보를 검색하고 응답합니다. 그 다음, 다른 질문을 해보겠습니다:

messages = [HumanMessage(content="What about in la?")]
thread = {"configurable": {"thread_id": "1"}}

for event in abot.graph.stream({"messages": messages}, thread):
    for v in event.values():
        print(v)

같은 thread_id를 사용함으로써, 에이전트는 이전 대화 맥락을 기억하고 있습니다.

8. 토큰 스트리밍

마지막으로, 실시간으로 AI의 응답을 받아보겠습니다:

from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver

memory = AsyncSqliteSaver.from_conn_string(":memory:")
abot = Agent(model, [tool], system=prompt, checkpointer=memory)

messages = [HumanMessage(content="What is the weather in SF?")]
thread = {"configurable": {"thread_id": "4"}}

async for event in abot.graph.astream_events({"messages": messages}, thread, version="v1"):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="|")

이 코드는 AI의 응답을 실시간으로 스트리밍하여 출력합니다.

결론

LangGraph를 사용하면 지속성과 스트리밍 기능을 갖춘 강력한 AI 에이전트를 구축할 수 있습니다. 이러한 기능들은 더 자연스럽고 효율적인 대화형 AI 시스템을 만드는 데 큰 도움이 됩니다.

이 튜토리얼을 통해 여러분도 자신만의 고급 AI 에이전트를 만들어 보시기 바랍니다. 질문이나 의견이 있다면 언제든 댓글로 남겨주세요. 행복한 코딩 되세요!

#LangChain

• [랭체인] 간단한 텍스트 생성 체인 만들기
• [랭체인] 텍스트 감성 분류 체인 만들기
• [랭체인] 간단한 챗봇 만들기
• [랭체인] 수학 연산 체인 구현하기
• [랭체인] SQL 데이터베이스 쿼리 자동화
• [랭체인] PythonREPL을 활용한 동적 코드 실행 챗봇 구현
• [랭체인] PDF 문서 기반 QA챗봇 구현
• [랭체인] 대화형 챗봇 만들기
• [랭체인] 도구 체인 구축 가이드
• [랭체인] 라우팅 체인 - 유연한 AI 응답 시스템 구축하기
• [랭체인] 간단한 ReAct 에이전트 구축하기
• [랭체인] 지능형 에이전트 구축하기
• [랭체인] Agentic Search: 더 스마트한 웹 검색 방법
• [랭체인] LangGraph를 사용하여 고급 대화형 AI 에이전트 구축하기
• [랭체인] LangGraph의 "Human in the Loop"를 활용하여 시스템을 구현하는 방법

_{Posted using Obsidian Steemit plugin}