Model
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4-1106-preview")
Tools
from typing import Annotated, List, Tuple, Union
from langchain.tools import BaseTool, StructuredTool, Tool
from langchain_experimental.tools import PythonREPLTool
from langchain_core.tools import tool
import random
python_repl_tool = PythonREPLTool()
@tool("lower_case", return_direct=False)
def to_lower_case(input:str) -> str:
"""Returns the input as all lower case."""
return input.lower()
@tool("random_number", return_direct=False)
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,python_repl_tool]
Helper Utils
LangChain 라이브러리를 사용하여, 사용자 지정 도구(agent, executor)를 통해 자연어 이해 및 생성 작업에 활용할 수 있는 대화형 에이전트를 생성
ChatOpenAI에서 가져온 LLM(Language Model)을 생성
create_openai_tools_agent 함수를 사용하여 지정된 LLM, 도구 리스트 그리고 시스템 프롬프트를 바탕으로 에이전트 생성AgentExecutor 에이전트와 함께 제공된 도구로 객체를 초기화agent_node() 주어진 상태를 기반으로 에이전트의 동작을 실행하고 결과 메시지를 반환from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
# 대화형 에이전트 생성
def create_agent(llm: ChatOpenAI, tools: list, system_prompt: str):
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
system_prompt,
),
MessagesPlaceholder(variable_name="messages"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
]
)
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
return executor
# 에이전트 실행 및 결과 메시지 반환
def agent_node(state, agent, name):
result = agent.invoke(state)
return {"messages": [HumanMessage(content=result["output"], name=name)]}
Create Agent Supervisor
members
시스템(슈퍼바이저)의 프롬프트 작성
프롬프트 템플릿 구성
작업 순서를 결정하는 체인을 생성
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
members = ["Lotto_Manager", "Coder"]
system_prompt = (
"You are a supervisor tasked with managing a conversation between the"
" following workers: {members}. Given the following user request,"
" respond with the worker to act next. Each worker will perform a"
" task and respond with their results and status. When finished,"
" respond with FINISH."
)
options = ["FINISH"] + members
function_def = {
"name": "route",
"description": "Select the next role.",
"parameters": {
"title": "routeSchema",
"type": "object",
"properties": {
"next": {
"title": "Next",
"anyOf": [
{"enum": options},
],
}
},
"required": ["next"],
},
}
prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder(variable_name="messages"),
(
"system",
"Given the conversation above, who should act next?"
" Or should we FINISH? Select one of: {options}",
),
]
).partial(options=str(options), members=", ".join(members))
supervisor_chain = (
prompt
| llm.bind_functions(functions=[function_def], function_call="route")
| JsonOutputFunctionsParser()
)
Create the AgnetState and Graph
AgentState는 에이전트의 상태 구조를 정의하는 딕셔너리 정의
앞에 선언한 Helper Utils 를 사용하여 ‘Lotto_Manager’, ‘Coder’ 멤버들에 대한 에이전트와 노드를 생성한다.
lotto
agent - 언어 모델(llm), 도구, 역할 설명을 매개변수 lotto_agent 객체를 생성
node - 에이전트 노드의 자세한 사항을 저장하는 lotto_node라는 부분 함수를 생성
code
lotto와 동일하게 agent, node 함수를 생성하는데 단지 다른 점은, 로컬에서 코드를 작성하기 위해 tools가 아닌 python_repl_tool를 사용하고 있다는 점이다.
AgentState 타입의 상태 그래프로 workflow를 생성
workflow에 앞에 선언했던 ‘Lotto_Manager’, ‘Coder’ 와 supervisor_chain을 추가
import operator
from typing import Annotated, Any, Dict, List, Optional, Sequence, TypedDict
import functools
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import StateGraph, END
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
next: str
lotto_agent = create_agent(llm, tools, "You are a senior lotto manager. you run the lotto and get random numbers")
lotto_node = functools.partial(agent_node, agent=lotto_agent, name="Lotto_Manager")
code_agent = create_agent(llm, [python_repl_tool], "You may generate safe python code to analyze data and generate charts using matplotlib.")
code_node = functools.partial(agent_node, agent=code_agent, name="Coder")
workflow = StateGraph(AgentState)
workflow.add_node("Lotto_Manager", lotto_node)
workflow.add_node("Coder", code_node)
workflow.add_node("supervisor", supervisor_chain)
Edges
반복문을 통해 각 멤버를 순회?하면서, 멤버와 'supervisor' 노드 사이에 경로를 추가
대화 중 다음 단계로 넘어갈 수 있는 선택지를 결정하는 규칙을 담은 conditional_map 생성
conditional_map에 조건을 포함하는 supervisor를 추가하고 시작점으로 세팅한다
그동안 생성했던 워크플로우를 컴파일한 최종 그래프를 생성한다
for member in members:
workflow.add_edge(member, "supervisor")
conditional_map = {k: k for k in members}
conditional_map["FINISH"] = END
workflow.add_conditional_edges("supervisor", lambda x: x["next"], conditional_map)
workflow.set_entry_point("supervisor")
graph = workflow.compile()
Run it
재귀 호출의 최대 제한을 20으로 설정한 후
데모 용 메시지()를 입력하고 출력을 기다린다.
그래프를 스트리밍하고 결과를 출력합니다.
graph.stream - 출력 결과는 "end" 키가 없는 상태만 출력됩니다.
final_response - 같은 메시지를 사용하여 그래프의 최종응답을 취득합니다.
최종 메시지는 출력 내용 중 두번째 메시지에 담겨있다.['messages'][1]
# 'stream'을 사용하여 단계별로 결과를 확인
config = {"recursion_limit": 20}
for s in graph.stream(
{
"messages": [
HumanMessage(content="무작위 로또 번호 10개를 가져와서 히스토그램에 10개의 빈으로 표시하고 마지막에 10개의 숫자가 무엇인지 알려주세요.")
]
}, config=config
):
if "__end__" not in s:
print(s)
print("----")
# 'invoke'를 사용하여 최종 결과를 얻음
final_response = graph.invoke(
{
"messages": [
HumanMessage(content="무작위 로또 번호 10개를 가져와서 히스토그램에 10개의 빈으로 표시하고 마지막에 10개의 숫자가 무엇인지 알려주세요.")
]
}, config=config
)
final_response['messages'][1].content