LangGraph - 실습 4. Hierarchical Agent Teams

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

 
환경 변수 로드

from dotenv import load_dotenv
load_dotenv() 

 

 

---

시작에 앞서 전체적인 구조를 살펴보자

어떠한 모양이 계속 복제하여, 결국 전체로 반복되는 일종의? 프랙탈(Fractal) 구조와 같이 작은 Agent Supervisor 구성이 반복? 적층되는 형태로(,, 어쩌면 레고 블록에 더 가까울 수도 있는) 그래프가 구성된다.

 

 

Create Tools

from typing import Annotated, List, Tuple, Union

import matplotlib.pyplot as plt
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langsmith import trace

tavily_tool = TavilySearchResults(max_results=5)

@tool
def scrape_webpages(urls: List[str]) -> str:
    """주어진 웹 페이지에서 requests와 bs4를 사용하여 상세 정보를 크롤링합니다."""
    loader = WebBaseLoader(urls)
    docs = loader.load()
    return "\\n\\n".join(
        [
            f'<Document name="{doc.metadata.get("title", "")}">\\n{doc.page_content}\\n</Document>'
            for doc in docs
        ]
    )
    
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Dict, Optional

from langchain_experimental.utilities import PythonREPL
from typing_extensions import TypedDict

_TEMP_DIRECTORY = TemporaryDirectory()
WORKING_DIRECTORY = Path(_TEMP_DIRECTORY.name)
print("WORKING_DIRECTORY :", WORKING_DIRECTORY)

@tool
def create_outline(
    points: Annotated[List[str], "주요 포인트 또는 섹션의 목록."],
    file_name: Annotated[str, "개요를 저장할 파일 경로."],
) -> Annotated[str, "저장된 개요 파일의 경로."]:
    """개요를 생성하고 저장합니다."""
    with (WORKING_DIRECTORY / file_name).open("w") as file:
        for i, point in enumerate(points):
            file.write(f"{i + 1}. {point}\\n")
    return f"개요가 {file_name}에 저장되었습니다."

@tool
def read_document(
    file_name: Annotated[str, "문서를 저장할 파일 경로."],
    start: Annotated[Optional[int], "시작 라인. 기본값은 0입니다"] = None,
    end: Annotated[Optional[int], "끝 라인. 기본값은 None입니다"] = None,
) -> str:
    """지정된 문서를 읽습니다."""
    with (WORKING_DIRECTORY / file_name).open("r") as file:
        lines = file.readlines()
    if start is not None:
        start = 0
    return "\\n".join(lines[start:end])

@tool
def write_document(
    content: Annotated[str, "문서에 작성될 텍스트 내용."],
    file_name: Annotated[str, "문서를 저장할 파일 경로."],
) -> Annotated[str, "저장된 문서 파일의 경로."]:
    """텍스트 문서를 만들고 저장합니다."""
    with (WORKING_DIRECTORY / file_name).open("w") as file:
        file.write(content)
    return f"문서가 {file_name}에 저장되었습니다."

@tool
def edit_document(
    file_name: Annotated[str, "편집할 문서의 경로."],
    inserts: Annotated[
        Dict[int, str],
        "키가 행 번호(1에서 시작)이고 값이 해당 행에 삽입될 텍스트인 사전.",
    ],
) -> Annotated[str, "편집된 문서 파일의 경로."]:
    """특정 행 번호에 텍스트를 삽입하여 문서를 편집합니다."""

    with (WORKING_DIRECTORY / file_name).open("r") as file:
        lines = file.readlines()

    sorted_inserts = sorted(inserts.items())

    for line_number, text in sorted_inserts:
        if 1 <= line_number <= len(lines) + 1:
            lines.insert(line_number - 1, text + "\\n")
        else:
            return f"오류: 행 번호 {line_number}이(가) 범위를 벗어났습니다."

    with (WORKING_DIRECTORY / file_name).open("w") as file:
        file.writelines(lines)

    return f"{file_name}에 편집된 문서를 저장했습니다."

repl = PythonREPL()

@tool
def python_repl(
    code: Annotated[str, "차트를 생성하기 위해 실행할 파이썬 코드."]
):
    """이것을 사용하여 파이썬 코드를 실행하세요. 값의 출력을 보려면 `print(...)`로 출력해야 합니다. 이것은 사용자에게 보이게 됩니다."""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"실행에 실패하였습니다. 에러: {repr(e)}"
    return f"성공적으로 실행되었습니다:\\n```python\\n{code}\\n```\\nStdout: {result}"

 

 

Helper Utils

from typing import Any, Callable, List, Optional, TypedDict, Union

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import Runnable
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI

from langgraph.graph import END, StateGraph

def create_agent(
    llm: ChatOpenAI,
    tools: list,
    system_prompt: str,
) -> str:
    """함수를 호출하는 에이전트를 생성하고 그래프에 추가합니다."""
    system_prompt += "\\n당신의 전문성에 따라 자율적으로 작업하며, 당신이 사용할 수 있는 도구를 사용합니다."
    " 명확성을 요구하지 마세요."
    " 당신의 다른 팀원들(그리고 다른 팀들)은 그들 자신의 전문성을 가진다고 발표."
    " 당신은 이유가 있어 선택되었습니다! 당신은 다음 팀원 중 한명입니다: {team_members}."
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                system_prompt,
            ),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]
    )
    agent = create_openai_functions_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)]}

def create_team_supervisor(llm: ChatOpenAI, system_prompt, members) -> str:
    """LLM 기반 라우터."""
    options = ["FINISH"] + members
    function_def = {
        "name": "route",
        "description": "다음 역할을 선택합니다.",
        "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",
                "위의 대화를 기준으로, 누가 다음으로 행동해야 할까요?"
                " 아니면 종료(FINISH)해야 할까요? 다음 중 하나를 선택하세요: {options}",
            ),
        ]
    ).partial(options=str(options), team_members=", ".join(members))
    return (
        prompt
        | llm.bind_functions(functions=[function_def], function_call="route")
        | JsonOutputFunctionsParser()
    )

 

 

Define Agent Teams

1) Research Team

import functools
import operator

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai.chat_models import ChatOpenAI
import functools

class ResearchTeamState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    team_members: List[str]
    next: str

llm = ChatOpenAI(model="gpt-4-1106-preview")

search_agent = create_agent(
    llm,
    [tavily_tool],
    "당신은 '타빌리' 검색 엔진을 이용해 최신 정보를 검색할 수 있는 연구 보조원입니다.",
)
search_node = functools.partial(agent_node, agent=search_agent, name="Search")

research_agent = create_agent(
    llm,
    [scrape_webpages],
    "당신은 'scrape_webpages' 함수를 이용하여 지정된 URL에서 더 자세한 정보를 스크랩할 수 있는 연구 보조원입니다.",
)
research_node = functools.partial(agent_node, agent=research_agent, name="Web_Scraper")

supervisor_agent = create_team_supervisor(
    llm,
    "당신은 다음 작업자들 사이의 대화를 관리하는 감독자입니다: Search, Web_Scraper. 각 작업자는 작업을 수행하고 그 결과와 상태를 보고합니다. 완료되면, FINISH를 응답하세요.",
    ["Search", "Web_Scraper"],
)

research_graph = StateGraph(ResearchTeamState)
research_graph.add_node("Search", search_node)
research_graph.add_node("Web_Scraper", research_node)
research_graph.add_node("supervisor", supervisor_agent)

research_graph.add_edge("Search", "supervisor")
research_graph.add_edge("Web_Scraper", "supervisor")
research_graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {"Search": "Search", "Web_Scraper": "Web_Scraper", "FINISH": END},
)

research_graph.set_entry_point("supervisor")
chain = research_graph.compile()

def enter_chain(message: str):
    results = {
        "messages": [HumanMessage(content=message)],
    }
    return results

research_chain = enter_chain | chain

# 연구 팀을 실행합니다
# print("--연구 팀을 실행합니다--")
# for s in research_chain.stream(
#     "영화'원더랜드'의 개봉일을 찾아서 한글로 답해주세요", {"recursion_limit": 100}
# ):
#     if "__end__" not in s:
#         print(s)
#         print("---")

 

2) Document Writing Team

import operator
from pathlib import Path

class DocWritingState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    team_members: str
    next: str
    current_files: str

def prelude(state):
    written_files = []
    if not WORKING_DIRECTORY.exists():
        WORKING_DIRECTORY.mkdir()
    try:
        written_files = [
            f.relative_to(WORKING_DIRECTORY) for f in WORKING_DIRECTORY.rglob("*")
        ]
    except:
        pass
    if not written_files:
        return {**state, "current_files": "작성된 파일이 없습니다."}
    return {
        **state,
        "current_files": "\\n아래는 팀이 디렉토리에 작성한 파일들입니다:\\n"
        + "\\n".join([f" - {f}" for f in written_files]),
    }

llm = ChatOpenAI(model="gpt-4-1106-preview")

doc_writer_agent = create_agent(
    llm,
    [write_document, edit_document, read_document],
    "당신은 연구 문서를 작성하는 전문가입니다.\\n"
    "아래는 현재 디렉토리에 있는 파일들입니다:\\n{current_files}",
)
context_aware_doc_writer_agent = prelude | doc_writer_agent
doc_writing_node = functools.partial(
    agent_node, agent=context_aware_doc_writer_agent, name="DocWriter"
)

note_taking_agent = create_agent(
    llm,
    [create_outline, read_document],
    "당신은 논문 개요 작성과 논문 작성을 위한 노트 정리를 담당하는 경험 많은 연구원입니다.{current_files}",
)

context_aware_note_taking_agent = prelude | note_taking_agent
note_taking_node = functools.partial(
    agent_node, agent=context_aware_note_taking_agent, name="NoteTaker"
)

chart_generating_agent = create_agent(
    llm,
    [read_document, python_repl],
    "You are a data viz expert tasked with generating charts for a research project."
    "{current_files}",
)
context_aware_chart_generating_agent = prelude | chart_generating_agent
chart_generating_node = functools.partial(
    agent_node, agent=context_aware_note_taking_agent, name="ChartGenerator"
)

doc_writing_supervisor = create_team_supervisor(
    llm,
    "당신은 다음 작업자들 사이의 대화를 관리하는 감독자입니다:  {team_members}. 각 작업자는 작업을 수행하고 그 결과와 상태를 보고합니다. 완료되면, FINISH를 응답하세요.",
    ["DocWriter", "NoteTaker", "ChartGenerator"],
)

authoring_graph = StateGraph(DocWritingState)
authoring_graph.add_node("DocWriter", doc_writing_node)
authoring_graph.add_node("NoteTaker", note_taking_node)
authoring_graph.add_node("ChartGenerator", chart_generating_node)
authoring_graph.add_node("supervisor", doc_writing_supervisor)

authoring_graph.add_edge("DocWriter", "supervisor")
authoring_graph.add_edge("NoteTaker", "supervisor")
authoring_graph.add_edge("ChartGenerator", "supervisor")

authoring_graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {
        "DocWriter": "DocWriter",
        "NoteTaker": "NoteTaker",
        "ChartGenerator": "ChartGenerator",
        "FINISH": END,
    },
)

authoring_graph.set_entry_point("supervisor")
chain = authoring_graph.compile()

def enter_chain(message: str, members: List[str]):
    results = {
        "messages": [HumanMessage(content=message)],
        "team_members": ", ".join(members),
    }
    return results

authoring_chain = (
    functools.partial(enter_chain, members=authoring_graph.nodes)
    | authoring_graph.compile()
)

#문서 작성 팀을 실행합니다.
# print("--문서 작성 팀을 실행합니다.--")
# for s in authoring_chain.stream(
#     "한글로, 시에 대한 개요를 작성한 후 디스크에 시를 작성하세요.",
#     {"recursion_limit": 100},
# ):
#     if "__end__" not in s:
#         print(s)
#         print("---")

 

 

Add Layers for.

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_openai.chat_models import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-1106-preview")

supervisor_node = create_team_supervisor(
    llm,
    "당신은 다음 팀들 사이의 대화를 관리하는 감독자입니다 : {team_members}. 주어진 사용자 요청에 대해, 다음으로 행동할 작업자를 지정하십시오. 각 작업자는 작업을 수행하고 그들의 결과와 상태를 보고합니다. 완료되면, FINISH를 응답하세요.", 
    ["Research team", "Paper writing team"],
)

class State(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next: str

def get_last_message(state: State) -> str:
    return state["messages"][-1].content

def join_graph(response: dict):
    return {"messages": [response["messages"][-1]]}

super_graph = StateGraph(State)
super_graph.add_node("Research team", get_last_message | research_chain | join_graph)
super_graph.add_node(
    "Paper writing team", get_last_message | authoring_chain | join_graph
)
super_graph.add_node("supervisor", supervisor_node)

super_graph.add_edge("Research team", "supervisor")
super_graph.add_edge("Paper writing team", "supervisor")
super_graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {
        "Paper writing team": "Paper writing team",
        "Research team": "Research team",
        "FINISH": END,
    },
)
super_graph.set_entry_point("supervisor")
super_graph = super_graph.compile()

 

 

Run it 

for s in super_graph.stream(
    {
        "messages": [
            HumanMessage(
                content="대한민국 인구 변화에 간략한 연구 내용울 차트를 포함한 한글 보고서로 작성하십시오."
            )
        ],
    },
    {"recursion_limit": 150},
):
    if "__end__" not in s:
        print(s)
        print("---")

* 페이지 링크 내용 : 

 

 

*참고; Colab print

{'supervisor': {'next': 'Research team'}}
---
{'Research team': {'messages': [HumanMessage(content='# Population Change in South Korea: Analysis and Trends\\n\\n## Abstract\\n\\nSouth Korea has experienced significant changes in its population dynamics over the past few decades. Factors such as declining birth rates, increased life expectancy, and migration patterns have all contributed to shifts in the demographic landscape. This research paper aims to provide an overview of the population change in South Korea, incorporating current data and trends.\\n\\n## Introduction\\n\\nSouth Korea, officially known as the Republic of Korea, has been witnessing a gradual shift in its demographic profile. The country\\'s economic development, urbanization, and changes in societal values have all played a role in altering population patterns. Understanding these changes is critical for policy planning and addressing future demographic challenges.\\n\\n## Methodology\\n\\nTo analyze population change in South Korea, we conducted a search using the tavily search engine to gather the most up-to-date and reliable data. Information was sourced from government statistics, demographic studies, and international databases to ensure a comprehensive understanding of the trends.\\n\\n## Results\\n\\n### Population Growth Rate\\n\\nSouth Korea\\'s population growth rate has been declining steadily. The country is experiencing one of the lowest birth rates in the world, leading to concerns about labor shortages and the economic implications of an aging population.\\n\\n### Age Structure\\n\\nThere is a notable shift towards an aging population, with a significant increase in the percentage of elderly citizens. This demographic trend is expected to continue, impacting social services and healthcare systems.\\n\\n### Urbanization\\n\\nUrban areas, particularly the Seoul metropolitan region, have seen population growth due to internal migration. This has led to challenges in housing, transportation, and urban planning.\\n\\n### Migration\\n\\nInternational migration has also affected South Korea\\'s population dynamics. The country has seen an influx of foreign workers and marriage migrants, which has implications for cultural integration and social policy.\\n\\n## Discussion\\n\\nThe population trends in South Korea reflect a broader pattern observed in many developed nations. The low birth rate and aging population pose challenges for economic growth and sustainability. Efforts to reverse these trends include government incentives for families, improvements in childcare and work-life balance, and policies to integrate foreign residents.\\n\\n## Conclusion\\n\\nSouth Korea\\'s population dynamics are changing, with significant implications for its future. Addressing these changes requires proactive policy measures and societal adaptation to ensure a stable and prosperous society.\\n\\n## References\\n\\n1. Korean Statistical Information Service (KOSIS)\\n2. United Nations Population Division\\n3. National Institute of Population and Social Security Research\\n\\n## Appendix: Wool Chart\\n\\nUnfortunately, as a research assistant, I do not have the capability to create a "wool chart." The term "wool chart" does not correspond to a known type of data visualization in demographic research. It is possible that there was a miscommunication or a typographical error. If you meant to refer to a "wall chart" or another form of data representation such as a bar graph or pie chart to visualize population data, please clarify, and I can proceed accordingly with the correct type of data visualization.', name='Search')]}}
---
{'supervisor': {'next': 'Paper writing team'}}
---
{'Paper writing team': {'messages': [HumanMessage(content='The research paper outline titled "Population Change in South Korea: Analysis and Trends" has been created and saved under the file name "Population_Change_in_South_Korea_Analysis." This document provides a structured framework for the analysis of demographic changes in South Korea, covering various aspects and implications. The specified sections include an Abstract, Introduction, Methodology, Results, Discussion, Conclusion, References, and an Appendix for data visualization clarification. The next steps include detailed information gathering, data analysis, and contextual discussion within demographic studies.', name='DocWriter')]}}
---
{'supervisor': {'next': 'Research team'}}
---
{'Research team': {'messages': [HumanMessage(content="Based on the search results, here is a summary of the demographic trends in South Korea that can be included in the **Introduction** section of your research paper:\\n\\n1. **Population Decline**: South Korea's population is experiencing a decline, with a decrease from 51,815,810 in 2022 to 51,741,963 in 2024. This trend of population decline has been consistent over the past few years.\\n\\n2. **Aging Population**: The median age in South Korea is 44.5 years as of 2023. There is an increasing elderly dependency ratio, indicating a growing share of the population aged 65 and above.\\n\\n3. **Low Birth Rate**: The fertility rate in South Korea has been low, with recent statistics showing a continual decrease in the number of births.\\n\\n4. **Urbanization**: South Korea has a high population density, and the urban population has been increasing, indicating continued urbanization within the country.\\n\\n5. **Changing Household Structures**: There is a rise in single-person households and a transformation in the traditional family structure.\\n\\n6. **Economic Implications**: As the economically active population changes, there are implications for labor force participation, employment rates, and economic dependency ratios.\\n\\n7. **Social Implications**: Changes in demographics are affecting social policies, including immigration and social change, as well as challenges related to the aging population.\\n\\n8. **Cultural and Policy Responses**: The South Korean government and society are responding to demographic challenges with various strategies and policies, which can be explored further in the research paper.\\n\\nThese trends provide an overview of the demographic situation in South Korea and set the stage for a detailed analysis in subsequent sections of your research paper. For a complete Introduction, you might consider elaborating on each of these points, providing a historical context, and discussing the significance of studying these trends.\\n\\nWould you like to explore any of these points in more detail, or shall we move on to another section of your research paper outline?", name='Search')]}}
---
{'supervisor': {'next': 'Paper writing team'}}
---
{'Paper writing team': {'messages': [HumanMessage(content='The introduction section for the research paper on demographic trends in South Korea has been created and saved as "Demographic_Trends_in_South_Korea_Introduction." \\n\\nWould you like to proceed with another section of the research paper or review and refine the existing content?', name='DocWriter')]}}
---
{'supervisor': {'next': 'Research team'}}
---
{'Research team': {'messages': [HumanMessage(content='The introduction section for the research paper on demographic trends in South Korea has been created and saved as "Demographic_Trends_in_South_Korea_Introduction." \\n\\nWould you like to proceed with another section of the research paper or review and refine the existing content?')]}}
---
{'supervisor': {'next': 'Research team'}}
---
{'Research team': {'messages': [HumanMessage(content='The introduction section for the research paper on demographic trends in South Korea has been created and saved as "Demographic_Trends_in_South_Korea_Introduction." \\n\\nWould you like to proceed with another section of the research paper or review and refine the existing content?')]}}
---
{'supervisor': {'next': 'Research team'}}
---
{'Research team': {'messages': [HumanMessage(content='The introduction section for the research paper on demographic trends in South Korea has been created and saved as "Demographic_Trends_in_South_Korea_Introduction." \\n\\nWould you like to proceed with another section of the research paper or review and refine the existing content?')]}}
---
{'supervisor': {'next': 'FINISH'}}
---

 

그런데.. 랭그래프 공식 사이트에 가보니.. 시각화되 결과를 확인할 수 있는 예제로 코드가 업데이트 되었다.. 
구 예제는 요정도로 마무리하고, 관련 내용을 추가로 업데이트 해야겠다.. >< 옴마- 빠르기도 해라 ㅋ 

https://github.com/langchain-ai/langgraphjs/blob/main/examples/multi_agent/hierarchical_agent_teams.ipynb

langgraphjs/examples/multi_agent/hierarchical_agent_teams.ipynb at main · langchain-ai/langgraphjs