LangGraph - 실습 2. Chat Executor

LangGraph - 2. 유형별 실습

intro
Agent Executor
👉 Chat Executor
Agent Supervisor
Hierarchical Agent Teams
Multi-agent Collaboration

 
[터미널] 환경 변수 및 깃헙 설정 
root 폴더에 환경변수 파일과 깃푸시 배재 파일을 생성한다. 

.env 작성
.gitignore 작성

 
[터미널] 가상환경 설정 및 실행
(옵션)가상 환경을 설정하고 실행한다( LangGraph_Agents는 개인적으로 작성한 임의의 이름)

설정 :  python -m venv LangGraph_Agents
실행(mac) :  source LangGraph_Agents/bin/activate

 
 

시작에 앞서 전체적인 구조를 살펴보자 
Chat Executor 그래프의 구조는 Agent Executor 그래프와 동일하다.  

Graph_Chat Executor

 
환경 변수 로드

from dotenv import load_dotenv
load_dotenv() 

 

State

상태관리 저장소 생성하기: State, 각 노드가 수행한 작업들을 기억(상태 기록)하는 기능
Agent Executor 예제와 달리 input, chat_history, agent_outcome, intermediate_steps 가 필요하지 않다. chat 형태이기 때문에 메타 데이터(operator.add) 및 채팅 내역 등을 intermediate_steps, chat_history 등이 아닌 chat(messages)에서 직접 관리한다.

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

 
 

Model

Agent Executor 예제와 달리 Agents 대신 Model을 구성한다. 이를 위해 'createOpenAI 함수 에이전트'가 아닌, langchain_openai에서 OpenAI의 채팅 모델을 위한 클래스 ChatOpenAI를 임포트한다. 

from langchain_openai import ChatOpenAI

model = ChatOpenAI(temperature=0, streaming=True)

 
 

Tools

커스텀 툴 생성하기 : to be 노드
앞의 예제에서 설명한 바와 같이 툴은 에이전트가 외부와 상호작용하는 과정에서 활용 할 수 있도록 해주는 인터페이스이고 Agent Executor 예제와 다른 점이 있다면, 앞에서 Model을 함수가 아닌 Class로 이미 선언했기 때문에 정의한 도구들을 리스트로 함께 묶어주는 작업까지만 한다. 

from langchain.tools import BaseTool, StructuredTool, Tool, tool
import random

@tool("lower_case", return_direct=True)
def to_lower_case(input:str) -> str:
  """Returns the input as all lower case."""
  return input.lower()

@tool("random_number", return_direct=True)
def random_number_maker(input:str) -> str:
    """Returns a random number between 0-100. input the word 'random'"""
    return random.randint(0, 100)

tools = [to_lower_case,random_number_maker]

 
 

Binding tools to models

 langchain의 도구를 OpenAI 채팅 기능에 포맷에 맞게 변환하는 과정으로 채팅 모델이 이 함수를 호출하여 사용할 수 있도록 위에서 정의한 도구들을 OpenAI 함수 포맷으로 변환하고, 그 함수들의 리스트를 생성한 후 이 함수들을 채팅 모델에 연결(bind)하는 과정이다.

from langchain.tools.render import format_tool_to_openai_function

functions = [format_tool_to_openai_function(t) for t in tools]
model = model.bind_functions(functions)

 

 

Nodes

노드 생성하기 : 노드
보유하고 있는 툴 목록, 에이전트 그리고 '조건부 엣지'에 사용될 로직을 정의한다. Agent Executor 예제와 다른 점은 call_agent 대신 call_model이 온 것과 call_tool 함수의 경우, 이전에는 data라는 딕셔너리에서 'agent_outcome'을 직접 가져와 tool_executor를 이용해 해당 도구를 호출하는 일종의 데이터의 바로가기 액션과 같은 역할을 수행했다면,  특정 상태 기반의 도구 호출에 초점을 맞추고 있다.
구체적으로 state라는 딕셔너리에서 메시지들을 가져온 다음 마지막 메시지에서 도구를 호출하기 위한 정보를 추출하여 ToolInvocation 객체를 생성한 후 tool_executor를 이용해 해당 도구를 실행하고, 그 응답을 포함하는 FunctionMessage를 생성하여 메시지 형태의 결과를 반환하게 된다.

## Nodes
from langchain_core.agents import AgentFinish
from langgraph.prebuilt import ToolInvocation
import json
from langchain_core.messages import FunctionMessage
from langgraph.prebuilt.tool_executor import ToolExecutor 

tool_executor = ToolExecutor(tools)
    
def call_model(state):
    messages = state['messages']
    response = model.invoke(messages)
    return {"messages": [response]} 

def call_tool(state):
    messages = state['messages']
    last_message = messages[-1] 
    action = ToolInvocation(
        tool=last_message.additional_kwargs["function_call"]["name"],
        tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]),
    )
    # print(f"The agent action is {action}")
    response = tool_executor.invoke(action)
    # print(f"The tool result is: {response}")
    function_message = FunctionMessage(content=str(response), name=action.tool)
    return {"messages": [function_message]}

def should_continue(state):
    messages = state['messages']
    last_message = messages[-1]
    if "function_call" not in last_message.additional_kwargs:
        return "end"
    else:
        return "continue"

 
 

Graph

그래프 정의하기 : Graph, 노드와 엣지의 모음
Agent Executor 예제와 비교하면 run_agent 대신 call_model이 들어간것 빼고는 동일하다. "agent"와 "action" 노드 간을 오가며 워크플로우를 실행하며 call_model과 call_tool 함수를 호출하는 애플리케이션으로 "agent"에서 시작해서 조건에 따라 "action"으로 이동하거나 종료하고, "action"에서는 다시 "agent"로 돌아갈 수 있는 순환 구조를 구성하는 과정이다.

from langgraph.graph import StateGraph, END
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("action", call_tool)

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)
workflow.add_edge('action', 'agent')

app = workflow.compile()

 

---

Run It with LangSmith

type 1 두 개의 툴을 모두 사용하는 경우

from langchain_core.messages import HumanMessage, SystemMessage

system_message = SystemMessage(content="you are a helpful assistant")
user_01 = HumanMessage(content="give me a random number and then write in words and make it lower case")
inputs = {"messages": [system_message,user_01]}

app.invoke(inputs)

 
type 2 하나의 툴만 사용하는 경우

from langchain_core.messages import HumanMessage, SystemMessage

system_message = SystemMessage(content="you are a helpful assistant")
user_01 = HumanMessage(content="plear write 'Merlion' in lower case")
inputs = {"messages": [system_message,user_01]}

app.invoke(inputs)

 
type 3 툴 사용이 필요없는 경우

from langchain_core.messages import HumanMessage, SystemMessage

system_message = SystemMessage(content="you are a helpful assistant")
user_01 = HumanMessage(content="what is a Merlion?")
inputs = {"messages": [system_message,user_01]}

app.invoke(inputs)