LangGraph는 LangChain 생태계의 확장입니다. LangChain은 여러 도구를 사용해 작업을 실행할 수 있는 AI 코딩 에이전트를 구축할 수 있지만, 여러 단계에 걸쳐 여러 체인이나 액터를 조정할 수는 없습니다. 이는 복잡한 작업을 수행하는 에이전트를 만들기 위한 중요한 동작입니다. LangGraph는 이러한 점을 염두에 두고 고안되었습니다. 에이전트 워크플로를 순환 그래프 구조로 취급하며, 각 노드는 함수 또는 Langchain 실행 가능한 객체를 나타내고 에지는 노드 간의 연결입니다.
LangGraph의 주요 기능은 다음과 같습니다.
Nodes: 도구와 같은 모든 함수 또는 Langchain 실행 가능한 객체(Runnable Object).
Edge: 노드 사이의 방향을 정의합니다.
Stateful Graph: 그래프의 기본 유형입니다. 노드를 통해 데이터를 처리할 때 상태 객체를 관리하고 업데이트하도록 설계되었습니다.
여러 에이전트 간의 조정을 통해 멀티 에이전트 시스템을 만드는 것이 목표라면 LangGraph를 사용하는 것이 좋습니다. 그러나 작업을 완료하기 위해 DAGs 또는 Chain을 생성하려는 경우에는 LangChain LCEL이 가장 적합합니다.
왜 LangGraph를 사용하나요?
LangGraph는 기존의 많은 솔루션을 개선할 수 있는 강력한 프레임워크입니다.
Improve RAG pipelines: LangGraph는 순환 그래프 구조로 RAG를 보강할 수 있습니다. 피드백 루프를 도입하여 검색된 객체의 품질을 평가하고, 필요한 경우 쿼리를 개선하고 프로세스를 반복할 수 있습니다.
Multi-Agent Workflows: LangGraph는 다중 에이전트 워크플로우를 지원하도록 설계되었습니다. 이는 복잡한 작업을 작은 하위 작업으로 나누어 해결하는 데 매우 중요합니다. 공유 상태와 서로 다른 LLM 및 도구를 가진 여러 에이전트가 협업하여 하나의 작업을 해결할 수 있습니다.
Human-in-the-loop: LangGraph는 휴먼 인더 루프 워크플로우를 기본적으로 지원합니다. 즉, 사람이 다음 노드로 이동하기 전에 상태를 검토할 수 있습니다.
Planning Agent: LangGraph는 LLM 플래너가 사용자 요청을 계획 및 분해하고, 실행자가 도구와 함수를 호출하며, LLM이 이전 결과물을 기반으로 답변을 합성하는 플래닝 에이전트를 구축하는 데 적합합니다.
Use Cases
복잡한 AI 코딩 에이전트가 도움이 될 수 있는 분야는 무궁무진합니다.
개인 에이전트: 텍스트, 음성, 제스처 등 사용자의 명령에 따라 작업을 도와주는 나만의 자비스 같은 비서가 전자 기기에 있다고 상상해 보세요. 이것이 바로 AI 에이전트의 가장 흥미로운 활용 사례 중 하나입니다!
AI 강사: 챗봇은 훌륭하지만 한계가 있습니다. 적절한 도구를 갖춘 AI 에이전트는 기본적인 대화 이상의 기능을 수행할 수 있습니다. 사용자 피드백에 따라 교육 방법을 조정할 수 있는 가상 AI 강사는 판도를 바꿀 수 있습니다.
소프트웨어 UX: AI 에이전트를 통해 소프트웨어의 사용자 경험을 개선할 수 있습니다. 에이전트는 애플리케이션을 수동으로 탐색하는 대신 음성이나 제스처 명령으로 작업을 수행할 수 있습니다.
공간 컴퓨팅: AR/VR 기술의 인기가 높아짐에 따라 AI 에이전트에 대한 수요도 증가할 것입니다. 에이전트는 주변 정보를 처리하고 필요에 따라 작업을 실행할 수 있습니다. 이는 곧 AI 에이전트의 최고의 사용 사례 중 하나가 될 것입니다.
LLM OS: 에이전트가 일등 시민이 되는 AI 우선 운영 체제. 에이전트는 일상적인 작업부터 복잡한 작업까지 담당하게 됩니다.
Key Features
LangGraph는 주기적인 상태 저장 다중 에이전트 시스템을 구축하기 위한 효율적인 프레임워크입니다. 이는 기존 LangChain 프레임워크의 공백을 메웁니다. LangChain의 확장이기 때문에 LangChain 생태계의 모든 좋은 점을 활용할 수 있습니다. LLM의 품질과 기능이 향상됨에 따라 복잡한 워크플로우를 자동화하기 위한 에이전트 시스템을 만드는 것이 훨씬 쉬워질 것입니다.
LangGraph는 순환적이고 상태 저장적인 멀티액터 에이전트 시스템을 구축할 수 있도록 해주는 LangChain의 확장입니다.
Nodes와 Edges가 있는 그래프 구조를 구현합니다. 노드는 함수 또는 도구이며, 에지는 노드와 노드 사이의 연결입니다.
Edges는 조건부 Edges와 일반 Edges의 두 가지 유형이 있습니다. 조건부 Edges는 한 Node에서 다른 Nodes로 이동하는 동안 조건을 가지며, 워크플로에 주기성을 추가하는 데 중요합니다.
순환형 멀티액터 에이전트를 구축하는 데는 LangGraph가 선호되는 반면, 체인이나 지시형 비순환 시스템을 만드는 데는 LangChain이 더 좋습니다.
LangGraph와 LangChain의 Agent 비교: Trade-off 관계
ReAct / LangChain
LangGraph
Reliability
LLM이 각 단계에서 올바른 결정을 내려야 하므로 신뢰성 저하
제어 흐름이 설정되어 있고 LLM이 각 노드에서 수행해야 할 특정 작업이 있으므로 더욱 안정적
Flexibility
LLM이 원하는 작업 순서를 선택할 수 있어 유연성 향상
각 노드에서 제어 흐름을 설정하면 동작이 제한되므로 유연성이 떨어짐
Compatibility with SLM
파라미터 사이즈가 적어지므로 호환성 저하
호환성 향상
LangGraph Basic 튜토리얼
Setup Environments
import os
from dotenv import load_dotenv
!echo "Your_OpenAI_Key" >> .env #openai key here
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
Define Tools
import pandas as pd
from langchain.tools import tool
@tool
def addition(x, y):
"""Addition of two number
:param: x: The first number to be added
:param: y: The second number to be added"""
return x+y
@tool
def subtraction(x, y):
"""Sumbtration of two number
:param: x: The first number the greater one
:param: y: The second number to be subtracted """
return x-y
@tool
def multiplication(x, y):
"""Multiplication of two number
:param: x: The first number to be multiplied
:param: y: The second number to be multiplied"""
return x*y
@tool
def division(x, y):
"""Division of two number
:param: x: The first number the greater one
:param: y: The second number to be devided """
return x/y
tools = [addition, subtraction, multiplication, division]
tool_dict = {} # this is going to be required during tool execution
for tool in tools:
tool_dict[tool.name]= tool
#%pip install langgraph
LLM & Planner
import json
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from typing import List, Optional
from langchain_openai import ChatOpenAI
from langchain.output_parsers.openai_tools import JsonOutputToolsParser
import operator
import os
from loguru import logger
class StrategyAgentState(TypedDict):
user_query: str
steps: Annotated[List, operator.add]
step_no: int
results: dict
final_response: str
end:bool
llm = ChatOpenAI(
model="gpt-3.5-turbo",
api_key=api_key
)
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="gpt-3.5-turbo"
)
def plan(state: StrategyAgentState):
"""The planner node, this is the brain of the system"""
user_question = state["user_query"]
steps = state["steps"]
results = state["results"]
end = state["end"]
if results is None:
# If result has not been populated yet we will start planning
SYSTEM_PROMT = "You are a helpful assitant who is good is mathematics.\
Do not calculate yourself let the tool do the calculation. Call one tool at a time"
prompt_template = ChatPromptTemplate.from_messages(
[("system", SYSTEM_PROMT),
("user", "{user_question}")])
planner = prompt_template | llm.bind_tools(tools)| JsonOutputToolsParser()
invoke_inputs = {"user_question": user_question}
steps = planner.invoke(invoke_inputs)
logger.info(f"Generated plans : {steps}")
return {'steps': steps}
elif results and not end:
# If result has been populated and end is not true we will go to end detector
SYSTEM_PROMT = "You need to decide whether a problem is solved or not. Just return ##YES if propblem is solved and ##NO \
if problem is not solved. Please expalain your reasoning as well. Make sure you use same template of ##YES and ##NO in final answer.\
Do not calculate yourself let the tool do the calculation"
prompt_template = ChatPromptTemplate.from_messages(
[("system", SYSTEM_PROMT),
("user", "{user_question}"),
("user", "{results}"),
("user", "{steps}")])
planner = prompt_template | llm
invoke_inputs = {"user_question": user_question, "steps":json.dumps(steps), "results":json.dumps(results)}
response = planner.invoke(invoke_inputs)
logger.info(f"End detector response : {response.content}")
if "##YES" in response.content:
return {'end': True}
elif "##NO" in response.content:
return {'end': False}
else:
# if end is not true and
SYSTEM_PROMT = "You are a helpful assitant who is good is mathematics.\
You are replanner assistant.\
If you are given previous steps and previous results. Do not start again. Call one function at a time.\
Do not calculate yourself let the tool do the calculation"
prompt_template = ChatPromptTemplate.from_messages(
[("system", SYSTEM_PROMT),
("user", "{user_question}"),
("user", "{steps}"),
("user", "{results}")])
planner = prompt_template | llm.bind_tools(tools)| JsonOutputToolsParser()
invoke_inputs = {"user_question": user_question, "steps":json.dumps(steps), "results":json.dumps(results)}
steps = planner.invoke(invoke_inputs)
logger.info(f"Pending plans : {steps}")
return {'steps': steps}
Tool Execution for Strategy Agent State
def tool_execution(state: StrategyAgentState):
""" Worker node that executes the tools of a given plan. Plan is json arguments
which can be sent to tools directly"""
steps = state["steps"]
step_no = state["step_no"] or 0
_results = state["results"] or {}
j= 0
for tool in steps[step_no: ]:
tool_name = tool['type']
args = tool["args"]
_results[tool_name+"_step_"+str(step_no+j)] = tool_dict[tool_name](args)
logger.info(f"{tool_name} is called with arguments {args}")
j=j+1
return {"results": _results, "step_no": step_no+j, }
def responder(state:StrategyAgentState):
user_question = state["user_query"]
results = state["results"]
SYSTEM_PROMT = "Generate final response by looking at the results and original user question."
prompt_template = ChatPromptTemplate.from_messages(
[("system", SYSTEM_PROMT),
("user", "{user_question}"),
("user", "{results}")])
model = prompt_template | llm
invoke_inputs = {"user_question": user_question, "results": json.dumps(results)}
response = model.invoke(invoke_inputs)
return {"final_response": response.content}
Strategy Agent State for Route
def route(state:StrategyAgentState):
"""A conditional route based on number of steps completed or end anounced by any other node,
this will either end the execution or will be sent to tools for planning"""
steps = state["steps"]
step_no = state["step_no"]
end = state["end"]
if end:
# We have executed all tasks
return "respond"
else:
# We are still executing tasks, loop back to the "tool" node
return "plan"
query = "what is 3 multiplied by 9 added to 45 then devide all by 6"
for s in agent.stream({"user_query": query}):
print(s)
print("--------------------")
