소개
[ 랭그래프 만드는 순서]
주요모듈 import
from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from IPython.display import Image, display # API 키를 환경변수로 관리하기 위한 설정 파일 from dotenv import load_dotenv # API 키 정보 로드 load_dotenv()상태(state)를 정의
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
우리가 정의하는 모든 것은 현재를 입력으로 받아 state를 업데이트하는 값을 반환합니다.
메시지를 직접 덮어쓰지 않고 현재 목록에 추가합니다. 이는 Annotated에 미리 빌드된 add_messages 함수를 통해 전달됩니다.
노드(node)를 추가
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini",temperature=0)
def chatbot(state: State):
return {"messages": [model.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
첫 번째 인수는 고유한 노드 이름입니다.
두 번째 인수는 언제든지 호출될 함수 또는 객체입니다.
edge 추가 (선 긋기)
graph_builder.add_edge(START, "chatbot") # 그래프를 실행할 때마다 작업을 시작할 위치를 알려줍니다 graph_builder.add_edge("chatbot", END) # 그래프에 "이 노드가 실행될 때마다 종료할 위치 지정정그래프를 컴파일
from IPython.display import Image, display
# 그래프 생성
graph = graph_builder.compile()
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
챗봇실행 test
while True:
user_input = input("User:")
if user_input. lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
for event in graph.stream({"messages": ("user", user_input)}):
for value in event.values():
print("user_input:",user_input)
print("Assistant:", value["messages"][-1].content)
위 그래프 예제의 챗봇은 메모리 기능이 없는 상태 임
[ 도구를 사용하여 챗봇의 기능 향상]
0. Tavily 검색엔진을 tool로 구성하기
챗봇이 "메모리에서" 답변할 수 없는 쿼리를 처리하기 위해 웹 검색 도구를 통합합니다.
봇은 이 도구를 사용하여 관련 정보를 찾고 더 나은 응답을 제공할 수 있습니다.
from langchain_community.tools.tavily_search import TavilySearchResults
tool = TavilySearchResults(max_results=2)
tools = [tool]
tool.invoke("랭그래프에서 노드란?")
1~4 단계 구현
# 정리된코드
from typing import Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from IPython.display import Image, display
# 1.상태설정
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
# 1.2 tool 설정
tool = TavilySearchResults(max_results=2)
tools = [tool]
llm = ChatOpenAI(model="gpt-4o-mini",temperature=0)
llm_with_tools = llm.bind_tools(tools)
# 2. 노드추가
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
# 3. edge 추가
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
# Any time a tool is called, we return to the chatbot to decide the next step
graph_builder.add_edge("tools", "chatbot")
graph_builder.set_entry_point("chatbot")
graph = graph_builder.compile()
# 그래프 시각화
display(Image(graph.get_graph().draw_mermaid_png()))
챗봇 실행
from langchain_core.messages import BaseMessage
i=0
while True:
user_input = input("User: ")
if user_input. lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
for event in graph.stream({"messages": [("user", user_input)]}):
for value in event.values():
i = i+ 1
print("###########values ", i, " :", value , "\n")
if isinstance(value["messages"][-1], BaseMessage):
print("Assistant:", value["messages"][-1].content, "\n")
print("#-----#", "\n")
결과와 배운 점
랭그 래프가 만만하지 않습니다. 우선은 랭체인부터 학습하길 추천 합니다
도움 받은 글 (옵션)
출처: https://langchain-ai.github.io/langgraph/tutorials/introduction/#part-1-build-a-basic-chatbot
출처2: https://www.youtube.com/watch?v=mvGp8Wz3KdI&t=240s