LangGraph - 실습 1. Agent Executor

LangGraph - 에이전트 유형별 실습

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

 
 

시작에 앞서 전체적인 구조를 살펴보자 
Agent Executor, 말 그대로 에이전트 실행자(? 기)이다. 아래는 생성한 그래프를 IPython.display를 통해 디스플레이한 도식이다. 
시작점과 종료시점이 선언되어 있고, 에이전트와 엑션는 루프 형태로 구성되어 있다. 그리고 도식을 통해, 무엇보다 중요한 '조건부 에지'가 에이전트에 설정되어 있음을 알 수 있다. 참고로 에이전트는 다양하게 구성될 수 있다. 

Agent Executor

 
 
환경 변수 로드

from dotenv import load_dotenv
load_dotenv() 

 

State

상태관리 저장소 생성하기 : State, 각 노드가 수행한 작업들을 기억(상태 기록)하는 기능

from typing import TypedDict, Annotated, List, Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
import operator

# GraphState - 각 노드가 수행한 작업들을 기억(상태 기록)하는 기능
class AgentState(TypedDict):
   input: str
   chat_history: list[BaseMessage] # 대화 내용 중 '이전 메시지' 목록
   agent_outcome: Union[AgentAction, AgentFinish, None] # 유효한 유형으로 `None`이 필요
   intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

 

Tools

커스텀 툴 생성하기 : to be 노드
툴은 에이전트가 외부와 상호작용하는 과정에서 활용 할 수 있도록 해주는 인터페이스이다. 때문에 에이전트(LLM)가 이해할 수 있도록 툴의 이름, 툴에 대한 설명, 어떻게 실행하는지, 어떤 입력을 받고 출력을 하는지 등에 대한 설명을 작성한다 

##   Custom Tools
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."""
    return random.randint(0, 100)

tools = [to_lower_case,random_number_maker]

random_value = random_number_maker.run('random')
SAM_value = to_lower_case.run('SAM') 

 

Agents

새로운 에이전트 설정하기: tobe노드
에이전트가 어떤 LLM Model을 사용하고 어떤 도구들(tools)을 사용할 수 있고 또한 어떤 컨디션(prompt)으로 대답할지 설정한다. 그리고 입력받을 값을 어떤 방식으로 운영(입력받을 값을 어떻게 처리할지, 챗 히스토리 저장공간설정, 중간스텝 저장공간설정)할 것인지를 등을 구성한다.

##    Agent - with new create_open_ai
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_openai.chat_models import ChatOpenAI

prompt = hub.pull("hwchase17/openai-functions-agent")
llm = ChatOpenAI(model="gpt-3.5-turbo-1106", streaming=True)
agent_runnable = create_openai_functions_agent(llm,tools,prompt)
inputs = {"input": "give me a random number and then write in words and make it lower case.",
          "chat_history": [],
          "intermediate_steps":[]}
agent_outcome = agent_runnable.invoke(inputs)

 

Nodes

노드 생성하기 : 노드
보유하고 있는 툴 목록, 에이전트 그리고 '조건부 엣지'에 사용될 로직을 정의한다. 

##      Nodes
from langchain_core.agents import AgentFinish
from langgraph.prebuilt.tool_executor import ToolExecutor

tool_executor = ToolExecutor(tools)

def run_agent(data):
    agent_outcome = agent_runnable.invoke(data)
    return {"agent_outcome": agent_outcome}

def execute_tools(data):
    agent_action = data['agent_outcome']
    output = tool_executor.invoke(agent_action)
    return {"intermediate_steps": [(agent_action, str(output))]}

def should_continue(data):
    if isinstance(data['agent_outcome'], AgentFinish):
        return "end"
    else:
        return "continue"

 

Graph

그래프 정의하기 : Graph, 노드와 엣지의 모음
앞서 상태관리를 위해 생성한 AgentState를 기반으로, 노드를 구성하고 그 안에 툴 또는 에이전트를 담는다. 그리고 시작점을 설정한 후 '조건부 엣지'와 함께 종료 시점을 정의한다. 그리고 마지막으로 이들을 엣지로 이어 그래프를 완성한다. 진짜 마지막으로 컴파일한다.  

##      Define the graph 
from langgraph.graph import END, StateGraph

workflow = StateGraph(AgentState)
workflow.add_node("agent", run_agent)
workflow.add_node("action", execute_tools)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "action",
        "end": END
    }
)
workflow.add_edge('action', 'agent')   # really this is the graph.
app = workflow.compile()    # to be like an app in here

 
 

---

Run It

 

스트림

## type 1 
inputs = {"input": "give me a random number and then write in words and make it lower case.", "chat_history": []}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

 
인보크

## type 2 
inputs = {"input": "give me a random number and then write in words and make it lower case", "chat_history": []}
output = app.invoke(inputs)
agent_outcome = output.get("agent_outcome").return_values['output']
# intermediate_steps = output.get("intermediate_steps"

 
인보크(도구 사용 X)

## type 3
inputs = {"input": "does it get cold in SF in Jan?", "chat_history": []}
output = app.invoke(inputs)
agent_outcome_no_Tools = output.get("agent_outcome").return_values['output']
print("\\n - agent_outcome_no_Tools :" ,agent_outcome_no_Tools)
# intermediate_steps_no_Tools = output.get("intermediate_steps")