[AI Agent] AI Agent with LangChain / LangGraph / LangSmith

 

[AI Agent ] AI Agent with LangChain / LangGraph / LangSmith

 

 

 

 

1. LangChain 이해하기¶

• https://wikidocs.net/book/14314

 

In [1]:

from IPython.core.display import display, HTML
display(HTML("<style>.container {width:90% !important;}</style>"))

 

 

/tmp/ipykernel_42131/3510566465.py:1: DeprecationWarning: Importing display from IPython.core.display is deprecated since IPython 7.14, please import from IPython.display
  from IPython.core.display import display, HTML

 

 

 

1) 기본 개념¶

 

 

(1) LangChain Expression Language(LCEL)¶

• 가장 기본적이고 일반적인 사용 사례는 prompt 템플릿과 모델을 함께 연결하는 것
• chain = prompt | model | output_parser

 

In [ ]:

from dotenv import load_dotenv
load_dotenv("/home1/irteamsu/data_ssd2/users/sjson/projects_NHN/llm.mcp/agents-from-scratch/.env")

 

In [ ]:

from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

 

In [ ]:

# 1. Prompt 정의 
# template 정의
template = "{country}의 수도는 어디인가요?"

# from_template 메소드를 이용하여 PromptTemplate 객체 생성
prompt_template = PromptTemplate.from_template(template)

# prompt 생성
prompt = prompt_template.format(country="대한민국")

prompt_template, prompt

 

In [ ]:

# 2. Model 정의 
model = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=2048,
    temperature=0.1,
    api_key="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"
)

 

In [ ]:

# 3. Output Parser 정의
output_parser = StrOutputParser()

 

In [ ]:

chain = prompt_template | model | output_parser

chain

 

In [ ]:

chain.get_graph() # 체인의 실행 그래프를 반환

 

 

LCEL은 다음과 같은 기본 구성 요소를 제공

• Runnable: 모든 LCEL 컴포넌트의 기본 클래스입니다.
• Chain: 여러 Runnable을 순차적으로 실행합니다.
• RunnableMap: 여러 Runnable을 병렬로 실행합니다.
• RunnableSequence: Runnable의 시퀀스를 정의합니다.
• RunnableLambda: 사용자 정의 함수를 Runnable로 래핑합니다.

 

 

(2) Runnable 프로토콜 - 사용자 정의 체인을 가능한 쉽게 만들 수 있도록!¶

• 사용자 정의 체인을 가능한 쉽게 만들 수 있도록, Runnable 프로토콜을 구현
• Runnable 프로토콜은 대부분의 컴포넌트에 구현되어 있음
• 이는 표준 인터페이스로, 다음의 메소드들이 포함됨
  • 동기 메소드
    • stream: 응답의 청크를 스트리밍합니다.
    • invoke: 입력에 대해 체인을 호출합니다.
    • batch: 입력 목록에 대해 체인을 호출합니다.
  • 비동기 메소드
    • astream: 비동기적으로 응답의 청크를 스트리밍합니다.
    • ainvoke: 비동기적으로 입력에 대해 체인을 호출합니다.
    • abatch: 비동기적으로 입력 목록에 대해 체인을 호출합니다.
    • astream_log: 최종 응답뿐만 아니라 발생하는 중간 단계를 스트리밍합니다.
  • +) 개념 정리
    • 동기 : 작업이 순차적으로 실행되기에, 이전 작업이 끝나야 다음 작업 수행
    • 비동기 : 작업을 요청한 후, 기다리지 않고 다음 작업을 실행
• 위에서 배운 Chain은 이런 여러 Runnable을 순차적으로 실행

 

In [ ]:

for token in chain.stream({"country": "미국"}):
    # 스트림에서 받은 데이터의 내용을 출력합니다. 줄바꿈 있게 출력하고, 버퍼를 즉시 비웁니다.
    print(token, end="\n", flush=True)

 

In [ ]:

chain.invoke({"country": "미국"})

 

In [ ]:

chain.batch([{"country": "미국"}, {"country": "한국"}])

 

 

(3) Runnable의 종류¶

• 기본 제공
  • RunnableMap: 여러 Runnable을 병렬로 실행합니다.
  • RunnableSequence: Runnable의 시퀀스를 정의합니다.
  • RunnableLambda: 사용자 정의 함수를 Runnable로 래핑합니다.
• 그 외
  • RunnablePassthrough 인스턴스: invoke() 메서드를 통해 입력된 데이터를 그대로 반환하거나 추가 키를 더해 전달 가능
    • 데이터를 변경하지 않고 파이프라인의 다음 단계로 전달하는 데 사용될 수 있음
  • RunnableParellel 클래스: 병렬로 실행 가능한 작업을 정의
  • RunnableWithMessageHistory : 특정 유형의 작업(체인)에 메시지 기록을 추가
    • 휘발성 저장
    • 영구 저장

 

In [ ]:

from langchain_core.runnables import RunnableParallel, RunnablePassthrough

runnable = RunnableParallel(
    # 전달된 입력을 그대로 반환하는 Runnable을 설정합니다.
    passed=RunnablePassthrough(),
    # 입력의 "num" 값에 3을 곱한 결과를 반환하는 Runnable을 설정합니다.
    extra=RunnablePassthrough.assign(mult=lambda x: x["num"] * 3),
    # 입력의 "num" 값에 1을 더한 결과를 반환하는 Runnable을 설정합니다.
    modified=lambda x: x["num"] + 1,
)

# {"num": 1}을 입력으로 Runnable을 실행합니다.
runnable.invoke({"num": 1})

 

In [ ]:

r = RunnablePassthrough.assign(mult=lambda x: x["num"] * 3)
r.invoke({"num": 1})

 

In [ ]:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

model = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=2048,
    temperature=0.1,
    api_key="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"
)
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "당신은 {ability} 에 능숙한 어시스턴트입니다. 20자 이내로 응답하세요",
        ),
        # 대화 기록을 변수로 사용, history 가 MessageHistory 의 key 가 됨
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),  # 사용자 입력을 변수로 사용
    ]
)
runnable = prompt | model  # 프롬프트와 모델을 연결하여 runnable 객체 생성

 

In [ ]:

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}  # 세션 기록을 저장할 딕셔너리


# 세션 ID를 기반으로 세션 기록을 가져오는 함수
def get_session_history(session_ids: str) -> BaseChatMessageHistory:
    print(session_ids)
    if session_ids not in store:  # 세션 ID가 store에 없는 경우
        # 새로운 ChatMessageHistory 객체를 생성하여 store에 저장
        store[session_ids] = ChatMessageHistory()
    return store[session_ids]  # 해당 세션 ID에 대한 세션 기록 반환


with_message_history = (
    RunnableWithMessageHistory(  # RunnableWithMessageHistory 객체 생성
        runnable,  # 실행할 Runnable 객체
        get_session_history,  # 세션 기록을 가져오는 함수
        input_messages_key="input",  # 입력 메시지의 키
        history_messages_key="history",  # 기록 메시지의 키
    )
)

 

In [ ]:

with_message_history.invoke(
    # 수학 관련 질문 "코사인의 의미는 무엇인가요?"를 입력으로 전달합니다.
    {"ability": "math", "input": "What does cosine mean?"},
    # 설정 정보로 세션 ID "abc123"을 전달합니다.
    config={"configurable": {"session_id": "abc123"}},
)

 

In [ ]:

with_message_history.invoke(
    # 능력과 입력을 설정합니다.
    {"ability": "math", "input": "이전의 내용을 한글로 답변해 주세요."},
    # 설정 옵션을 지정합니다.
    config={"configurable": {"session_id": "abc123"}},
)

 

In [ ]:

# 새로운 session_id로 인해 이전 대화 내용을 기억하지 않습니다.
with_message_history.invoke(
    # 수학 능력과 입력 메시지를 전달합니다.
    {"ability": "math", "input": "이전의 내용을 한글로 답변해 주세요"},
    # 새로운 session_id를 설정합니다.
    config={"configurable": {"session_id": "def234"}},
)

 

 

(4) RunnableSequence : Runnable instance 묶기¶

 

In [ ]:

from langchain_core.runnables import RunnableLambda, RunnableSequence

def add_one(x: int) -> int:
    return x + 1

def mul_two(x: int) -> int:
    return x * 2

runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence1 = runnable_1 | runnable_2
sequence1.invoke(1)

 

In [ ]:

# Or equivalently:
sequence2 = RunnableSequence(first=runnable_1, last=runnable_2)
sequence2.invoke(1)

 

 

(5) Runtime Arguments Binding¶

• 때로는 Runnable 시퀀스 내에서 Runnable을 호출할 때, 이전 Runnable의 출력이나 사용자 입력에 포함되지 않은 상수 인자를 전달해야 할 경우가 있음
  • 이때 Runnable.bind()를 사용하면 인자를 쉽게 전달할 수 있음

 

In [ ]:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            # 대수 기호를 사용하여 다음 방정식을 작성한 다음 풀이하세요.
            "Write out the following equation using algebraic symbols then solve it. "
            "Use the format\n\nEQUATION:...\nSOLUTION:...\n\n",
        ),
        (
            "human",
            "{equation_statement}",  # 사용자가 입력한 방정식 문장을 변수로 받습니다.
        ),
    ]
)

model = ChatOpenAI(
    model="gpt-4o-mini",
    max_tokens=2048,
    temperature=0.1,
    api_key="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"
)

# 방정식 문장을 입력받아 프롬프트에 전달하고, 모델에서 생성된 결과를 문자열로 파싱합니다.
runnable = (
    {"equation_statement": RunnablePassthrough()} | prompt | model | StrOutputParser()
)

# 예시 방정식 문장을 입력하여 결과를 출력합니다.
print(runnable.invoke("x raised to the third plus seven equals 12"))

 

In [ ]:

runnable = (
    # 실행 가능한 패스스루 객체를 생성하여 "equation_statement" 키에 할당합니다.
    {"equation_statement": RunnablePassthrough()}
    | prompt  # 프롬프트를 파이프라인에 추가합니다.
    | model.bind(
        stop="SOLUTION"
    )  # 모델을 바인딩하고 "SOLUTION" 토큰에서 생성을 중지하도록 설정합니다.
    | StrOutputParser()  # 문자열 출력 파서를 파이프라인에 추가합니다.
)
# "x raised to the third plus seven equals 12"라는 입력으로 파이프라인을 실행하고 결과를 출력합니다.
print(runnable.invoke("x raised to the third plus seven equals 12"))

 

 

• binding 의 특히 유용한 활용 방법 중 하나는 호환되는 OpenAI 모델에 OpenAI Functions 를 연결하는 것임

 

In [ ]:

openai_function = {
    "name": "solver",  # 함수의 이름
    # 함수의 설명: 방정식을 수립하고 해결합니다.
    "description": "Formulates and solves an equation",
    "parameters": {  # 함수의 매개변수
        "type": "object",  # 매개변수의 타입: 객체
        "properties": {  # 매개변수의 속성
            "equation": {  # 방정식 속성
                "type": "string",  # 방정식의 타입: 문자열
                "description": "The algebraic expression of the equation",  # 방정식의 대수식 표현
            },
            "solution": {  # 해답 속성
                "type": "string",  # 해답의 타입: 문자열
                "description": "The solution to the equation",  # 방정식의 해답
            },
        },
        "required": ["equation", "solution"],  # 필수 매개변수: 방정식과 해답
    },
}

 

In [ ]:

# 다음 방정식을 대수 기호를 사용하여 작성한 다음 해결하세요.
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Write out the following equation using algebraic symbols then solve it.",
        ),
        ("human", "{equation_statement}"),
    ]
)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key="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").bind(
    function_call={"name": "solver"},  # openai_function schema 를 바인딩합니다.
    functions=[openai_function],
)
runnable = {"equation_statement": RunnablePassthrough()} | prompt | model
# x의 세제곱에 7을 더하면 12와 같다
runnable.invoke("x raised to the third plus seven equals 12")

 

 

2) Messages¶

메시지 타입	설명
SystemMessage	시스템 지시나 프롬프트를 정의할 때 사용됩니다. LLM의 행동이나 문체를 제어할 수 있습니다. 예: "You are a helpful assistant."
HumanMessage	사람이 보낸 메시지를 나타냅니다. 예: "What's the weather like today?"
AIMessage	LLM이 생성한 응답을 나타냅니다. ***내부적으로 role="assistant"역할** . 예: "The weather is sunny today."
FunctionMessage	함수 호출 후 반환된 응답 메시지입니다. tool calling 이후에 LLM이 응답을 계속 이어나갈 수 있도록 합니다.
ToolMessage	LangChain에서 tool 결과를 전달할 때 사용하는 메시지입니다. 예를 들어 외부 API, DB 질의, 내부 계산 등 다양한 도구의 결과를 LLM에 전달 FunctionMessage와 유사하지만 더 범용적입니다.
ChatMessage(role=..., content=...)	커스텀 역할(role)을 갖는 일반 메시지입니다. tool_call_id와 연결되어 있어 어떤 호출에 대한 응답인지 명확히 구분 가능 예: role="user" 혹은 "moderator" 등 임의 지정 가능

• Tool Calling하는 경우, 핵심 흐름
  1. 사용자 질문 → HumanMessage
  2. LLM 응답 (Tool 호출 지시 포함) → AIMessage(tool_calls=[...])
  3. Tool 실행 결과 응답 → ToolMessage(tool_call_id, content)
  4. LLM이 최종 응답 → AIMessage(content="...")

 

In [ ]:

from langchain.schema import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="Tell me a joke."),
    AIMessage(content="Why don't scientists trust atoms? Because they make up everything!")
]

 

 

3) Agent란?¶

• 에이전트는 사전에 정의된 규칙이나 명시적인 프로그래밍 없이도 스스로 결정을 내리고 행동
  • 구성
    • AI Model
    • Capabilities and Tools
• LangChain에서 에이전트는 다음과 같은 구성요소로 이루어져 있습니다:
  • Agent: 의사 결정을 담당하는 핵심 컴포넌트입니다.
  • Tools: 에이전트가 사용할 수 있는 기능들의 집합입니다.
  • Toolkits: 관련된 도구들의 그룹입니다.
  • AgentExecutor: 에이전트의 실행을 관리하는 컴포넌트입니다.

 

 

(1) Tool & Tool Binding¶

• Tool이 가져야 하는 것
  • 함수가 수행하는 작업에 대한 텍스트 설명
  • Arguments with typings
  • (Optional) Outputs with typings
• Tool 실행 방법
  1. LLM에게 툴의 존재에 대해 가르치고 필요할 때 텍스트 기반 호출을 생성하도록 지시하
  2. LLM은 call weather_tool('Paris')와 같은 도구 호출을 나타내는 텍스트를 생성하도록 함
  3. 에이전트는 이 응답을 읽고, 툴 호출이 필요한지 식별하고, LLM을 대신하여 툴을 실행
• Tool 주는 방법
  • system prompt을 통해

 

In [ ]:

import re
import requests
from bs4 import BeautifulSoup
from langchain.agents import tool


# 도구를 정의합니다.
@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)


@tool
def add_function(a: float, b: float) -> float:
    """Adds two numbers together."""
    return a + b


@tool
def naver_news_crawl(news_url: str) -> str:
    """Crawls a 네이버 (naver.com) news article and returns the body content."""
    # HTTP GET 요청 보내기
    response = requests.get(news_url)

    # 요청이 성공했는지 확인
    if response.status_code == 200:
        # BeautifulSoup을 사용하여 HTML 파싱
        soup = BeautifulSoup(response.text, "html.parser")

        # 원하는 정보 추출
        title = soup.find("h2", id="title_area").get_text()
        content = soup.find("div", id="contents").get_text()
        cleaned_title = re.sub(r"\n{2,}", "\n", title)
        cleaned_content = re.sub(r"\n{2,}", "\n", content)
    else:
        print(f"HTTP 요청 실패. 응답 코드: {response.status_code}")

    return f"{cleaned_title}\n{cleaned_content}"


tools = [get_word_length, add_function, naver_news_crawl]

 

In [ ]:

from langchain_openai import ChatOpenAI

# 모델 생성
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key="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")

# 도구 바인딩
llm_with_tools = llm.bind_tools(tools)

 

In [ ]:

llm_with_tools.invoke("What is the length of the word '바보'?").tool_calls

 

In [ ]:

from langchain_core.output_parsers.openai_tools import JsonOutputToolsParser

# 도구 바인딩 + 도구 파서
chain = llm_with_tools | JsonOutputToolsParser(tools=tools)

# 실행 결과
tool_call_results = chain.invoke("What is the length of the word '바보'?")

 

In [ ]:

print(tool_call_results, end="\n\n==========\n\n")
# 첫 번째 도구 호출 결과
single_result = tool_call_results[0]
# 도구 이름
print(single_result["type"])
# 도구 인자
print(single_result["args"])

 

 

• 위와 같이 llm_with_tools 와 JsonOutputToolsParser 를 연결하여 tool_calls 를 parsing 하여 결과를 확인
  • chain에 function calling하는 함수를 추가해준다면 function까지 자동 실행

 

In [ ]:

def execute_tool_calls(tool_call_results):
    """
    도구 호출 결과를 실행하는 함수

    :param tool_call_results: 도구 호출 결과 리스트
    :param tools: 사용 가능한 도구 리스트
    """
    # 도구 호출 결과 리스트를 순회합니다.
    for tool_call_result in tool_call_results:
        # 도구의 이름과 인자를 추출합니다.
        tool_name = tool_call_result["type"]
        tool_args = tool_call_result["args"]

        # 도구 이름과 일치하는 도구를 찾아 실행합니다.
        # next() 함수를 사용하여 일치하는 첫 번째 도구를 찾습니다.
        matching_tool = next((tool for tool in tools if tool.name == tool_name), None)

        if matching_tool:
            # 일치하는 도구를 찾았다면 해당 도구를 실행합니다.
            result = matching_tool.invoke(tool_args)
            # 실행 결과를 출력합니다.
            print(f"[실행도구] {tool_name}\n[실행결과] {result}")
        else:
            # 일치하는 도구를 찾지 못했다면 경고 메시지를 출력합니다.
            print(f"경고: {tool_name}에 해당하는 도구를 찾을 수 없습니다.")

chain = llm_with_tools | JsonOutputToolsParser(tools=tools) | execute_tool_calls

chain.invoke("What is the length of the word '바보'?")

 

 

(2) Agent와 AgentExecutor¶

• AI 에이전트 워크플로(AI Agent Workflow) - Thought-Action-Observation
  1. Thought: The LLM part of the Agent decides what the next step should be.
     • Internal Reasoning
     • Updated thought
       • 반영: Reflecting
       • 최종 조치 Final Action
  2. Action: The agent takes an action, by calling the tools with the associated arguments.
     • Tool Usage
  3. Observation: The model reflects on the response from the tool.
     • Feedback from the Environment:
       • This observation is then added to the prompt as additional context

 

 

• tool을 선택하는 llm를 구현하는 방법
  • 방법 1) 위처럼 ChatOpenAI()에 .bind_tools 직접 하고 invoke할 수도 있지만, [ChatOpenAI().bind_tools() + .invoke()]
    • → "툴을 사용할 수 있는 LLM"을 만들고 직접 호출
    • 구조: ChatOpenAI 모델에 tools 정보를 바인딩해서 툴 콜링 가능한 LLM 객체를 생성.
    • 역할: tool 호출은 가능하지만, 멀티스텝 추론, 결정적 워크플로우, 에이전트 실행 로직 없음
  • 방법 2) LangChain에서 제공하는 Agent 형태로 불러올 수 있음 !! (예를 들어 create_tool_calling_agent() + agent_executor.invoke()를 통해서, 종류는 여기)
    • → "툴 사용 포함 추론을 관리하는 에이전트"를 구성하고 실행
    • 구조: LLM + 툴 + 추론 관리 체계를 갖춘 LangChain Agent 생성
    • 역할: 툴 사용 여부 판단 → 툴 호출 → 결과 반영하여 최종 응답 생성까지 전체 흐름 관리
    • 다중 툴, 조건부 사용, 반복 툴 호출 등 복잡한 시나리오 대응 가능

 

In [ ]:

import re
import requests
from bs4 import BeautifulSoup
from langchain.agents import tool


# 도구를 정의합니다.
@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)


@tool
def add_function(a: float, b: float) -> float:
    """Adds two numbers together."""
    return a + b


@tool
def naver_news_crawl(news_url: str) -> str:
    """Crawls a 네이버 (naver.com) news article and returns the body content."""
    # HTTP GET 요청 보내기
    response = requests.get(news_url)

    # 요청이 성공했는지 확인
    if response.status_code == 200:
        # BeautifulSoup을 사용하여 HTML 파싱
        soup = BeautifulSoup(response.text, "html.parser")

        # 원하는 정보 추출
        title = soup.find("h2", id="title_area").get_text()
        content = soup.find("div", id="contents").get_text()
        cleaned_title = re.sub(r"\n{2,}", "\n", title)
        cleaned_content = re.sub(r"\n{2,}", "\n", content)
    else:
        print(f"HTTP 요청 실패. 응답 코드: {response.status_code}")

    return f"{cleaned_title}\n{cleaned_content}"


one_tool = [get_word_length]
tools = [get_word_length, add_function, naver_news_crawl]

 

 

[1] 톺아보기¶

• 방법 1
  • ChatOpenAI().bind_tools() 하면 출력되는 RunnableBinding
    • ✅ RunnableBinding은 LLM과 함께 사용하는 함수들 (tool definitions) 을 LLM에 묶어서 실행하는 Runnable입니다.
    • LLM + tools → 묶어서 하나의 실행 단위로 만든 것 = RunnableBinding
    • 내부 정의 및 동작 과정

          class RunnableBinding(Runnable):
              def __init__(self, bound, kwargs):
                  self.bound = bound        # LLM 등 Runnable
                  self.kwargs = kwargs      # tools 같은 추가 매개변수
      

      1. RunnableBinding.invoke(input)이 호출되면 bound에 kwargs를 바인딩해서 새로운 Runnable을 만듦
      2. 그 새 runnable에 input을 넘겨 실행
  • 특징
    • → "툴을 사용할 수 있는 LLM"을 만들고 직접 호출
    • 구조: ChatOpenAI 모델에 tools 정보를 바인딩해서 툴 콜링 가능한 LLM 객체를 생성.
    • 역할: tool 호출은 가능하지만, 멀티스텝 추론, 결정적 워크플로우, 에이전트 실행 로직 없음

 

In [ ]:

llm_OpenAI = ChatOpenAI(model="gpt-4-turbo", api_key='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', temperature=0)
llm_OpenAI

 

In [ ]:

llm_OpenAI.bind_tools(tools)

 

 

• 방법 2
  • Agent + AgentExecutor
    • Agent :
      • 구성
        • RunnableAssign → ChatPromptTemplate → RunnableBinding으로 묶인 ChatOpenAI(model+tools) → ToolsAgentOutputParser
          1. RunnableAssign : 메시지 전처리나 구성 요소 재조합을 위한 유틸 체인 - RunnableAssign(mapper={...}): 특정 key를 추가하는 역할 - 여기선 agent_scratchpad를 추가함 - 지금까지 agent가 호출한 Tool들의 이력과 결과를 텍스트 형태로 정리해, 다음 프롬프트에 다시 포함시키는 용도 - RunnableLambda(...): lambda 함수로 직접 가공 - 이 경우엔 message_formatter를 통해 intermediate_steps를 포맷팅하여 agent_scratchpad로 넘김 - intermediate_steps은 이전 툴 호출 기록

      •     RunnableAssign(mapper={
            agent_scratchpad: RunnableLambda(lambda x: format_log_to_str(x['intermediate_steps']))
            })
            | PromptTemplate(input_variables=['agent_scratchpad', 'input'], input_types={}, partial_variables={'tools': 'get_word_length(word: str) -> int - Returns the length of a word.\nadd_function(a: float, b: float) -> float - Adds two numbers together.\nnaver_news_crawl(news_url: str) -> str - Crawls a 네이버 (naver.com) news article and returns the body content.', 'tool_names': 'get_word_length, add_function, naver_news_crawl'}, metadata={'lc_hub_owner': 'hwchase17', 'lc_hub_repo': 'react', 'lc_hub_commit_hash': 'd15fe3c426f1c4b3f37c9198853e4a86e20c425ca7f4752ec0c9b0e97ca7ea4d'}, template='Answer the following questions as best you can. You have access to the following tools:\n\n{tools}\n\nUse the following format:\n\nQuestion: the input question you must answer\nThought: you should always think about what to do\nAction: the action to take, should be one of [{tool_names}]\nAction Input: the input to the action\nObservation: the result of the action\n... (this Thought/Action/Action Input/Observation can repeat N times)\nThought: I now know the final answer\nFinal Answer: the final answer to the original input question\n\nBegin!\n\nQuestion: {input}\nThought:{agent_scratchpad}')
            | RunnableBinding(bound=ChatOpenAI(client=<openai.resources.chat.completions.completions.Completions object at 0x7f3077271870>, async_client=<openai.resources.chat.completions.completions.AsyncCompletions object at 0x7f3077240610>, root_client=<openai.OpenAI object at 0x7f3077287580>, root_async_client=<openai.AsyncOpenAI object at 0x7f30772718d0>, model_name='gpt-4-turbo', temperature=0.0, model_kwargs={}, openai_api_key=SecretStr('**********')), kwargs={'stop': ['\nObservation']}, config={}, config_factories=[])
            | ReActSingleInputOutputParser()
        

      • 종류
        • LangChain Agent에서도 주로 create_react_agent()와 create_tool_calling_agent()를 통해 만들어지는 agent를 가장 대표적으로 씀
          • create_react_agent() : ReAct prompting을 사용하는 agent를 만든다 (하지만 Multi-Input tool 불가능 ㅇㅇ) - Intended Model Type : LLM - Supports Chat History : O - Supports Multi-Input Tools : X - Supports Parallel Function Calling : X
          • create_tool_calling_agent(): 모델이 하나 이상의 도구(tool)를 호출해야 할 상황을 감지하고, 그 도구에 전달할 입력값들을 구조화된 형태(예: JSON)로 응답할 수 있도록 하는 기능 - 25.05.08 현재까지 아래의 세 가지 기능 모두 지원하는 유일한 agent
            - Intended Model Type : Chat - Supports Chat History : O - Supports Multi-Input Tools : O - Supports Parallel Function Calling : O
    • AgentExecutor : Agent와 Tool을 실행해주는 메인 컨트롤러
      • 내부 루프invoke() → iterate() → 내부 루프: plan() → execute_tool() → 결과 전달 → plan()... until AgentFinish
      • 주요 인자
        • agent: 실행 루프의 각 단계에서 계획을 생성하고 행동을 결정하는 에이전트
        • tools: 에이전트가 사용할 수 있는 유효한 도구 목록
        • return_intermediate_steps: 최종 출력과 함께 에이전트의 중간 단계 경로를 반환할지 여부
        • max_iterations: 실행 루프를 종료하기 전 최대 단계 수
        • early_stopping_method: 에이전트가 AgentFinish를 반환하지 않을 때 사용할 조기 종료 방법. ("force" or "generate")

 

In [ ]:

# 1. `create_react_agent()` 방식 
#     - prompt를 보면 위에서 언급한 방식처럼 Thought, Action, Observation, Thought의 반복으로 이뤄짐을 알 수 있다 
#     - 논문 : https://arxiv.org/abs/2210.03629
from langchain import hub

prompt = hub.pull("hwchase17/react")
prompt.pretty_print()

 

In [ ]:

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent

# Choose the LLM to use
llm = ChatOpenAI(model="gpt-4-turbo", api_key='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', temperature=0)

# Construct the ReAct agent
agent = create_react_agent(llm, tools, prompt)
agent
# ReActSingleInputOutputParser인 거 보면 위 설명대로 multi input tool들은 처리하지 못하는 것을 확인할 수 있다 

 

In [ ]:

from langchain.agents import AgentExecutor

# Create an agent executor by passing in the agent and tools
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
agent_executor

 

In [ ]:

tool_prompt = hub.pull("hwchase17/openai-tools-agent")
tool_prompt.pretty_print()

 

In [ ]:

from langchain.agents import create_tool_calling_agent

tool_agent = create_tool_calling_agent(llm_OpenAI, tools, prompt)
tool_agent
# 위의 create_react_agent와 output_parser 부분이 다르다 아마 json 형식으로 반환받겠지 

 

In [ ]:

tool_agent_executor = AgentExecutor(agent=tool_agent, tools=tools, verbose=True)
tool_agent_executor

 

 

[2] 예제¶

 

In [ ]:

from langchain.tools import tool
from typing import List, Dict, Annotated
from langchain_experimental.utilities import PythonREPL
import requests
from bs4 import BeautifulSoup
from typing import List, Dict

class GoogleNews:
    def __init__(self):
        self.base_url = "https://news.google.com"

    def search_by_keyword(self, query: str, k: int = 5) -> List[Dict[str, str]]:
        """구글 뉴스에서 query로 검색 후, 상위 k개의 뉴스 제목과 링크 반환"""
        search_url = f"{self.base_url}/search?q={query}"
        headers = {"User-Agent": "Mozilla/5.0"}
        response = requests.get(search_url, headers=headers)

        if response.status_code != 200:
            return [{"title": "뉴스 검색 실패", "url": search_url}]

        soup = BeautifulSoup(response.text, "html.parser")
        article_links = soup.select("article h3 a")[:k]

        results = []
        for a in article_links:
            title = a.get_text(strip=True)
            href = a.get("href", "")
            url = self.base_url + href[1:] if href.startswith(".") else href
            results.append({"title": title, "url": url})

        return results


# 도구 생성
@tool
def search_news(query: str) -> List[Dict[str, str]]:
    """Search Google News by input keyword"""
    news_tool = GoogleNews()
    return news_tool.search_by_keyword(query, k=5)


# 도구 생성
@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """Use this to execute python code. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    result = ""
    try:
        result = PythonREPL().run(code)
    except BaseException as e:
        print(f"Failed to execute. Error: {repr(e)}")
    finally:
        return result


print(f"도구 이름: {search_news.name}")
print(f"도구 설명: {search_news.description}")
print(f"도구 이름: {python_repl_tool.name}")
print(f"도구 설명: {python_repl_tool.description}")

 

In [ ]:

# tools 정의
tools = [search_news, python_repl_tool]

 

In [ ]:

from langchain_core.prompts import ChatPromptTemplate

# 프롬프트 생성
# 프롬프트는 에이전트에게 모델이 수행할 작업을 설명하는 텍스트를 제공합니다. (도구의 이름과 역할을 입력)
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful assistant. "
            "Make sure to use the `search_news` tool for searching keyword related news.",
        ),
        ("placeholder", "{chat_history}"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

 

In [ ]:

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent

# LLM 정의
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Agent 생성
agent = create_tool_calling_agent(llm, tools, prompt)

 

In [ ]:

from langchain.agents import AgentExecutor

# AgentExecutor 생성
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    max_execution_time=10,
    handle_parsing_errors=True,
)

# AgentExecutor 실행
result = agent_executor.invoke({"input": "AI 투자와 관련된 뉴스를 검색해 주세요."})

print("Agent 실행 결과:")
print(result["output"])

 

 

(+) 개념 설명¶

 

 

[1]. Thought¶

• Thought의 예시
• | Type of Thought | Example | | --------------- | ------- | | 계획  Planning | "이 작업을 1) 데이터 수집, 2) 추세 분석, 3) 보고서 생성의 세 단계로 나누어야 합니다."
  “I need to break this task into three steps: 1) gather data, 2) analyze trends, 3) generate report” | | 분석  Analysis | "오류 메시지에 따르면 데이터베이스 연결 매개 변수에있는 것 같습니다."
  “Based on the error message, the issue appears to be with the database connection parameters” | | 의사 결정  Decision Making | "사용자의 예산 제약을 감안할 때 중간 계층 옵션을 추천해야 합니다."
  “Given the user’s budget constraints, I should recommend the mid-tier option” | | 문제 해결  Problem Solving | "이 코드를 최적화하려면 먼저 프로파일링하여 병목 현상을 식별해야 합니다."
  “To optimize this code, I should first profile it to identify bottlenecks” | | 메모리 통합  Memory Integration | "사용자가 이전에 Python에 대한 선호도를 언급했으므로 Python으로 예제를 제공하겠습니다."
  “The user mentioned their preference for Python earlier, so I’ll provide examples in Python” | | 자기 성찰  Self-Reflection | "마지막 접근 방식이 잘 작동하지 않았으므로 다른 전략을 시도해야 합니다."
  “My last approach didn’t work well, I should try a different strategy” | | 목표 설정  Goal Setting | "이 작업을 완료하려면 먼저 수락 기준을 설정해야 합니다."
  “To complete this task, I need to first establish the acceptance criteria” | | 우선순위 지정  Prioritization | "새로운 기능을 추가하기 전에 보안 취약점을 해결해야 합니다."
  “The security vulnerability should be addressed before adding new features” |

* ToolFormer \& ReAct와 같은 논문이 여기에 해당됨

1. Toolformer: Language Models Can Teach Themselves to Use Tools¶

• 출처: Meta AI (2023)
• 핵심 아이디어: 기존 LLM은 도구(API, 계산기 등)를 능동적으로 사용할 수 없었습니다. Toolformer는 LLM이 스스로 학습 데이터를 생성하여 API 호출 위치와 방법을 학습할 수 있도록 합니다.
• 방법:
  1. 기존 LLM에 다양한 API (예: Wikipedia 검색, 계산기 등)를 호출하게 하고,
  2. 그 결과를 포함한 문장을 생성 → 기존 문장보다 더 좋은 예측 확률을 보이는 API 호출만 학습 데이터로 채택
  3. 그 데이터를 기반으로 LLM을 미세조정 (Fine-tuning)
• 결과:
  • 외부 도구(계산기, 날짜 계산, 번역기, 검색 API 등)를 능동적으로 활용
  • Prompt 설계 없이도 도구 호출 능력이 내재화됨
• 의의:
  • 인간의 어노테이션 없이도 API 사용 능력을 학습
  • Agent 시스템에 유용한 foundation 기술

🧠 2. ReAct: Synergizing Reasoning and Acting in Language Models¶

• 출처: Princeton & Google DeepMind (2022)
• 핵심 아이디어: LLM이 단순하게 텍스트를 생성하는 것이 아니라, 문제 해결을 위해 사고와 행동을 번갈아 수행하게 함. 예: 생각하고, 검색하고, 다시 생각하고, 답을 도출하는 방식.
• 방법:
  • 다음과 같은 형식의 Prompt를 설계:

  Thought: 지금 무엇을 해야 할지 생각  
  Action: 도구 사용 (예: 검색, 계산 등)  
  Observation: 도구 사용 결과  
  Thought: 결과를 기반으로 다음 단계 생각  
  ...
  Final Answer: 최종 답
  

• 예시:
• Q: Where was the 2022 World Cup held? Thought: I need to find the host country of the 2022 World Cup. Action: Search["2022 World Cup host country"] Observation: The 2022 World Cup was held in Qatar. Thought: Now I know the answer. Final Answer: Qatar
• 결과:
  • 정보 검색 기반 QA, 수학 계산, 게임 탐색 등에서 우수한 성능
  • Multi-hop reasoning과 tool interaction의 통합
• 의의:
  • LLM이 수동적 텍스트 생성자가 아니라, 행동을 통해 능동적 문제 해결자로 진화할 수 있는 기반
  • 현재 Agent 시스템의 prompt 구조에 많은 영향을 줌 (AutoGPT, LangChain, AutoGen 등)

✅ 비교 요약¶

항목	Toolformer	ReAct
목표	외부 API 도구를 스스로 학습	사고와 행동의 동시 수행
학습 방식	Self-supervised 도구 호출 학습	Reasoning & Action prompt chaining
특징	도구 사용을 능동적으로 삽입	각 단계별 사고과정 명시적 표현
사용 예	계산기, 번역기, 검색 등 API	웹 검색, 계산, 게임, 다단계 추론
응용처	Foundation Model, Agent 튜닝	Interactive QA, Multi-hop 문제 해결

[2]. Actions¶

• Agent의 종류
• | Type of Agent | Description | | ------------- | ----------- | | JSON 에이전트  JSON Agent | 수행할 작업은 JSON 형식으로 지정됩니다.
  The Action to take is specified in JSON format. | | 코드 에이전트  Code Agent | 에이전트는 외부에서 해석되는 코드 블록을 작성합니다.
  The Agent writes a code block that is interpreted externally. | | Function-calling 에이전트  Function-calling Agent | 각 작업에 대한 새 메시지를 생성하도록 미세 조정된 JSON 에이전트의 하위 범주입니다.
  It is a subcategory of the JSON Agent which has been fine-tuned to generate a new message for each action. |
• Action의 종류 | Type of Action | Description | | -------------- | ----------- | | 정보 수집  Information Gathering | 웹 검색 수행, 데이터베이스 쿼리 또는 문서 검색.
  Performing web searches, querying databases, or retrieving documents. | | 도구 사용  Tool Usage | API 호출, 계산 실행 및 코드 실행.
  Making API calls, running calculations, and executing code. | | 환경 상호 작용  Environment Interaction | 디지털 인터페이스를 조작하거나 물리적 장치를 제어합니다.
  Manipulating digital interfaces or controlling physical devices. | | 통신  Communication | 채팅을 통해 사용자와 소통하거나 다른 상담원과 협업합니다.
  Engaging with users via chat or collaborating with other agents. |

구현 방안 1. The Stop and Parse Approach - JSON 형식 파싱¶

구현 방안 2. Code Agent¶

[3]. Observations¶

• 피드백 수집: 작업이 성공적이었는지(또는 실패했는지)에 대한 데이터나 확인을 수신합니다.
• 결과 추가: 새로운 정보를 기존 맥락에 통합하여 기억을 효과적으로 업데이트합니다.
• 전략 조정: 업데이트된 맥락을 활용하여 후속적인 생각과 행동을 다듬습니다.

 

 

2. LangGraph¶

 

 

1) 기본 개념 - State, Node, Edge, 그리고 StateGraph¶

• LangGraph의 구성
  1. State
     • 그래프의 모든 노드와 엣지에 대한 입력 스키마 역할
     • 주로 TypedDict나 Pydantic BaseModel을 사용하여 정의
  2. Node
     • LangGraph에서 실제 작업을 수행하는 단위
     • 각 노드는 특정 기능을 수행하는 Python 함수로 구현
     • Node가 할 수 있는 기능들
       • LLM calls: Generate text or make decisions
       • Tool calls: Interact with external systems
       • Conditional logic: Determine next steps
       • Human intervention: Get input from users
  3. Edge
     • LangGraph에서 노드 간의 연결을 나타냄
     • 그래프의 실행 흐름을 결정하는 중요한 요소
     • START로 시작하여 END로 끝남
     • add_conditional_edges("node1", condition_function, {"type1":"node2", ... })를 통해 조건부 라우팅 가능!

=> 그리고 이걸 하나의 그래프 객체로 묶을 수 있는 게 바로 StateGraph이다

• 그래프 정의 후, 컴파일 후 실행 !

 

In [ ]:

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: str

 

In [ ]:

from langgraph.graph import StateGraph, START, END

graph = StateGraph(State)

 

In [ ]:

def node_1(state):
    print("---Node 1---")
    return {"graph_state": state['graph_state'] +" I am"}

def node_2(state):
    print("---Node 2---")
    return {"graph_state": state['graph_state'] +" happy!"}

def node_3(state):
    print("---Node 3---")
    return {"graph_state": state['graph_state'] +" sad!"}

 

In [ ]:

graph.add_node("node_1", node_1)
graph.add_node("node_2", node_2)
graph.add_node("node_3", node_3)

 

In [ ]:

from langgraph.prebuilt import ToolNode, tools_condition
from langchain.tools import tool
from typing import List, Dict, Annotated
from langchain_experimental.tools.python.tool import PythonAstREPLTool

import requests
from bs4 import BeautifulSoup
from typing import List, Dict

class GoogleNews:
    def __init__(self):
        self.base_url = "https://news.google.com"

    def search_by_keyword(self, query: str, k: int = 5) -> List[Dict[str, str]]:
        """구글 뉴스에서 query로 검색 후, 상위 k개의 뉴스 제목과 링크 반환"""
        search_url = f"{self.base_url}/search?q={query}"
        headers = {"User-Agent": "Mozilla/5.0"}
        response = requests.get(search_url, headers=headers)

        if response.status_code != 200:
            return [{"title": "뉴스 검색 실패", "url": search_url}]

        soup = BeautifulSoup(response.text, "html.parser")
        article_links = soup.select("article h3 a")[:k]

        results = []
        for a in article_links:
            title = a.get_text(strip=True)
            href = a.get("href", "")
            url = self.base_url + href[1:] if href.startswith(".") else href
            results.append({"title": title, "url": url})

        return results


# 도구 생성
@tool
def search_news(query: str) -> List[Dict[str, str]]:
    """Search Google News by input keyword"""
    news_tool = GoogleNews()
    return news_tool.search_by_keyword(query, k=5)

@tool
def python_code_interpreter(code: str):
    """Call to execute python code."""
    return PythonAstREPLTool().invoke(code)

# 도구 리스트 생성
tools = [search_news, python_code_interpreter]

# ToolNode 초기화
tool_node = ToolNode(tools)

 

In [ ]:

import random
from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:
    
    # Often, we will use state to decide on the next node to visit
    user_input = state['graph_state'] 
    
    # Here, let's just do a 50 / 50 split between nodes 2, 3
    if random.random() < 0.5:

        # 50% of the time, we return Node 2
        return "node_2"
    
    # 50% of the time, we return Node 3
    return "node_3"

 

In [ ]:

graph.add_edge(START, "node_1")
graph.add_conditional_edges("node_1", decide_mood)
graph.add_edge("node_2", END)
graph.add_edge("node_3", END)

 

In [ ]:

app = graph.compile()
result = app.invoke({'graph_state':'안녕'})
result

 

 

2) 체크포인터 & 서브 그래프¶

(1) 체크포인터 - 상태 기억하기¶

• LangGraph에서 상태를 저장하고 관리하는 핵심 기능
• 이를 통해 구현할 수 있는 기능
  • 메모리 : MemorySaver
    • LangChain의 RunnableWithMessageHistory와 같이 휘발성 저장, 영구 저장 모두 가능
  • 오류 복구
  • 인간-인-더-루프 (HITL) 상호작용

 

In [ ]:

# 이전 대화 참조 예시
def reference_previous_info(state):
    user_name = state.values.get('user_name')
    if user_name:
        return f"{user_name}님, 이전에 말씀하신 내용을 기억하고 있습니다."
    return "죄송합니다, 이전 정보를 찾을 수 없습니다."

graph.add_node("reference_info", reference_previous_info)

 

In [ ]:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from typing import TypedDict
from langgraph.errors import NodeInterrupt

# 상태 정의
class State(TypedDict):
    messages: list
    context: str

# MemorySaver 인스턴스 생성
memory = MemorySaver()

# 그래프 생성 및 컴파일
workflow = StateGraph(State)
# ... 노드 및 엣지 정의 ...
graph = workflow.compile(checkpointer=memory)

# 그래프 실행
config = {"configurable": {"thread_id": "user_123"}}
input_data = {"messages": [{"role": "user", "content": "안녕하세요"}], "context": ""}
result = graph.invoke(input_data, config=config)

# 상태 검사
state = graph.get_state(config)
print("현재 상태:", state.values)

 

In [ ]:

try:
    result = graph.invoke(input_data, config=config)
except Exception as e:
    print(f"오류 발생: {e}")
    last_stable_state = graph.get_state(config)
    graph.update_state(config, last_stable_state.values)
    print("마지막 안정 상태로 복구됨")

 

In [ ]:

# HITL 예시
def human_review_needed(state):
    if state.values.get('confidence_score', 0) < 0.7:
        raise NodeInterrupt("인간 검토 필요")
    return state

workflow.add_node("check_confidence", human_review_needed)
graph = workflow.compile(checkpointer=memory, interrupt_before=["check_confidence"])

try:
    result = graph.invoke(input_data, config=config)
except NodeInterrupt:
    human_input = input("검토 후 결정을 입력하세요: ")
    graph.update_state(config, {"human_decision": human_input})
    result = graph.invoke(None, config=config)  # 실행 재개

 

In [ ]:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph
from typing import TypedDict

class State(TypedDict):
    messages: list
    context: str

memory = MemorySaver()
workflow = StateGraph(State)
# ... 노드 및 엣지 정의 ...
graph = workflow.compile(checkpointer=memory)

 

In [ ]:

config1 = {"configurable": {"thread_id": "user_1"}}
graph.invoke({"messages": [{"role": "user", "content": "안녕하세요"}], "context": ""}, config=config1)

saved_state = graph.get_state(config1)
print("User 1 저장된 상태:", saved_state.values)

 

In [ ]:

config2 = {"configurable": {"thread_id": "user_2"}}
graph.invoke({"messages": [{"role": "user", "content": "다른 사용자입니다"}], "context": ""}, config=config2)

user2_state = graph.get_state(config2)
print("User 2 현재 상태:", user2_state.values)

 

In [ ]:

graph.update_state(config1, saved_state.values)
graph.update_state(config1, {"context": "이전 대화: 인사"})

 

 

(2) Sub-graph¶

• 큰 그래프 내에서 독립적으로 실행되는 작은 그래프

 

In [ ]:

# 메인 그래프 구현 

from typing import Optional, TypedDict, Literal 
from langgraph.graph import StateGraph, START, END
from IPython.display import Image
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
import operator

class MainState(TypedDict):
   messages: Annotated[List[AnyMessage], operator.add]  # 메시지 히스토리
   context: str  # 컨텍스트 정보
   subgraph_result: Optional[str]
   processing_status: str

def preprocessing(state: MainState) -> MainState:
    """데이터 전처리를 수행하는 노드"""
    return {
        "context": f"Context from: {state['messages'][-1].content}",
        "processing_status": "preprocessing_complete"
    }

def postprocessing(state: MainState) -> MainState:
    """서브그래프 실행 결과를 후처리하는 노드"""
    context = state.get("context", "")
    return {
        "subgraph_result": f"Final result based on context: {context}",
        "processing_status": "complete"
    }


def route_next(state: MainState) -> Literal["postprocessing", "reprocess"]:
    """다음 단계를 결정하는 라우터"""
    if state["processing_status"] == "preprocessing_complete":
        return "postprocessing"
    return "reprocess"

# 메인 그래프 구성
main_graph = StateGraph(MainState)

 

In [ ]:

# 서브 그래프 구성 
from typing import Annotated, List
from typing_extensions import TypedDict
import operator

from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END

class SubGraphState(TypedDict):
    """서브그래프의 상태를 정의하는 클래스"""
    messages: Annotated[List[AnyMessage], operator.add]  # 메시지 히스토리
    context: str  # 컨텍스트 정보
def process_node(state: SubGraphState):
    """입력 메시지를 처리하는 노드"""
    current_message = state["messages"][-1]
    # 메시지 처리 로직
    processed_result = f"Processed: {current_message.content}"
    return {"context": processed_result}

def respond_node(state: SubGraphState):
    """응답을 생성하는 노드"""
    context = state["context"]
    # 응답 생성 로직
    response = AIMessage(content=f"Response based on: {context}")
    return {"messages": [response]}

# 서브그래프 인스턴스 생성
subgraph = StateGraph(SubGraphState)

# 노드 추가
subgraph.add_node("process", process_node)
subgraph.add_node("respond", respond_node)

# 엣지 추가
subgraph.add_edge(START, "process")      # 시작 -> 처리
subgraph.add_edge("process", "respond")  # 처리 -> 응답
subgraph.add_edge("respond", END)        # 응답 -> 종료

# 그래프 컴파일
compiled_subgraph = subgraph.compile()

 

In [ ]:

# 메인 그래프 노드 추가
main_graph.add_node("preprocessing", preprocessing)
main_graph.add_node("subgraph", subgraph.compile())  # 기존 서브그래프
main_graph.add_node("postprocessing", postprocessing)

# 메인 그래프 엣지 추가
main_graph.add_edge(START, "preprocessing")
main_graph.add_edge("preprocessing", "subgraph")
main_graph.add_conditional_edges(
    "subgraph", 
    route_next,
    {
        "postprocessing": "postprocessing",
        "reprocess": "preprocessing"
    }
)
main_graph.add_edge("postprocessing", END)

 

In [ ]:

# 그래프 컴파일
compiled_maingraph = main_graph.compile()

# 그래프 시각화
maingraph_image = compiled_maingraph.get_graph(xray=True).draw_mermaid_png()
display(Image(maingraph_image))

 

 

+) Agent와 WorkFlow 구분하기¶

• 에이전트를 "agentic systems"로 정의하며, 이를 두 가지로 구분합니다:
  • 워크플로우(Workflows): 사전 정의된 코드 경로를 따라 LLM과 도구를 조율하는 시스템입니다.
  • 에이전트(Agents): LLM이 자체적으로 프로세스와 도구 사용을 동적으로 지시하며, 작업 수행 방식을 스스로 결정하는 시스템입니다.
• 워크플로우는 예측 가능성과 일관성을 제공하며, 에이전트는 유연성과 모델 기반 의사결정이 필요한 경우에 적합합니다.
• 해당 내용은 아래 링크 참고
  • https://langchain-ai.github.io/langgraph/tutorials/workflows/#orchestrator-worker
  • https://www.anthropic.com/engineering/building-effective-agents

 

 

3. 실전 예제¶

 

 

1) LangGraph에 MCP 더하기 (with LangChain MCP Adapter)¶

 

In [1]:

from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini", api_key="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")

 

In [9]:

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition

# 여러 개의 MCP 호환 툴 서버들을 동시에 실행하거나 연결하여, 해당 툴들을 LangChain에서 사용할 수 있도록 래핑해주는 클라이언트를 생성
async with MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            # Make sure to update to the full absolute path to your math_server.py file
            "args": ["/home1/irteamsu/data_ssd2/users/sjson/projects/mcp_test/server/math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            "command": "python",
            # Make sure to update to the full absolute path to your math_server.py file
            "args": ["/home1/irteamsu/data_ssd2/users/sjson/projects/mcp_test/server/weather_server.py"],
            "transport": "stdio",

            # make sure you start your weather server on port 8000
            # "url": "http://localhost:8000/sse",
            # "transport": "sse",
        }
    }
) as client:
    
    tools = client.get_tools()
    def call_model(state: MessagesState):
        response = model.bind_tools(tools).invoke(state["messages"])
        return {"messages": response}

    builder = StateGraph(MessagesState)
    builder.add_node(call_model)
    builder.add_node(ToolNode(tools))
    builder.add_edge(START, "call_model")
    builder.add_conditional_edges(
        "call_model",
        tools_condition, #  langchain.prebuilt에 있는 함수 아래의 route_tools 함수랑 동일한 역할을 한다 (내부 동작 확인 가능)
    )
    builder.add_edge("tools", "call_model")
    graph = builder.compile()
    math_response = await graph.ainvoke({"messages": "what's (3 + 5) x 12?"})
    weather_response = await graph.ainvoke({"messages": "what is the weather in nyc?"})

 

In [10]:

# def route_tools(
#     state: State,
# ):
#     """
#     마지막 메시지에 도구 호출이 있으면 ToolNode로 라우팅하기 위해 conditional_edge에서 사용한다.
#     그렇지 않으면 종료로 라우팅한다.
#     """
#     if isinstance(state, list):
#         ai_message = state[-1]
#     elif messages := state.get("messages", []):
#         ai_message = messages[-1]
#     else:
#         raise ValueError(f"No messages found in input state to tool_edge: {state}")
#     if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
#         return "tools" # 도구를 사용해야 할 때 "tools"를 반환한다.
#     return END

 

In [5]:

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

 

 

 

 

2) Multi-Tool Agent 구현하기¶

 

In [ ]:

from langchain_openai import ChatOpenAI
from langchain.tools import tool

model = ChatOpenAI(model="gpt-4o-mini", api_key="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")

# setup the simple tools using LangChain tool decorator
@tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b


@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


@tool
def square(a: int) -> int:
    """Calculates the square of a number."""
    a = int(a)
    return a * a

# setup the toolkit
toolkit = [add, multiply, square]

 

In [ ]:

from langchain.agents import create_tool_calling_agent

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

#define system prompt for tool calling agent
system_prompt = """ You are a mathematical assistant.
        Use your tools to answer questions. If you do not have a tool to
        answer the question, say so. """

tool_calling_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        MessagesPlaceholder("chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder("agent_scratchpad"),
    ]
)

tool_runnable = create_tool_calling_agent(model, toolkit, prompt  = tool_calling_prompt)
tool_actions = tool_runnable.invoke({"input": "hi! what's 1+1 and  2 times 2", 'intermediate_steps': []})

 

In [ ]:

def run_tool_agent(state):
    agent_outcome = tool_runnable.invoke(state)

    #this agent will overwrite the agent outcome state variable
    return {"agent_outcome": agent_outcome}

 

In [ ]:

from langgraph.prebuilt.tool_node import ToolNode

# tool executor invokes the tool action specified from the agent runnable
# they will become the nodes that will be called when the agent decides on a tool action.

tool_executor = ToolNode(toolkit)

# Define the function to execute tools
# This node will run a different tool as specified in the state variable agent_outcome
def execute_tools(state):
    # Get the most recent agent_outcome - this is the key added in the `agent` above
    agent_action = state['agent_outcome']
    if type(agent_action) is not list:
        agent_action = [agent_action]
    steps = []
    #sca only returns an action while tool calling returns a list
    # convert single actions to a list
    
    for action in agent_action:
    # Execute the tool
        output = tool_executor.invoke(action)
        print(f"The agent action is {action}")
        print(f"The tool result is: {output}")
        steps.append((action, str(output)))
    # Return the output
    return {"intermediate_steps": steps}

 

In [ ]:

from typing import TypedDict, Annotated,Union
from langchain_core.agents import AgentAction, AgentFinish
from langchain.agents.output_parsers.tools import ToolAgentAction
from langchain_core.messages import BaseMessage
import operator


class AgentState(TypedDict):
   # The input string from human
   input: str
   # The list of previous messages in the conversation
   chat_history: list[BaseMessage]
   # The outcome of a given call to the agent
   # Needs 'list' as a valid type as the tool agent returns a list.
   # Needs `None` as a valid type, since this is what this will start as
   # this state will be overwritten with the latest everytime the agent is run
   agent_outcome: Union[AgentAction, list, ToolAgentAction, AgentFinish, None]

   # List of actions and corresponding observations
   # These actions should be added onto the existing so we use `operator.add` 
   # to append to the list of past intermediate steps
   intermediate_steps: Annotated[list[Union[tuple[AgentAction, str], tuple[ToolAgentAction, str]]], operator.add]

 

In [ ]:

def should_continue(data):
    # If the agent outcome is an AgentFinish, then we return `exit` string
    # This will be used when setting up the graph to define the flow
    if isinstance(data['agent_outcome'], AgentFinish):
        return "END"
    # Otherwise, an AgentAction is returned
    # Here we return `continue` string
    # This will be used when setting up the graph to define the flow
    else:
        return "CONTINUE"

 

In [ ]:

from langgraph.graph import END, StateGraph

# Define a new graph
workflow = StateGraph(AgentState)

# When nodes are called, the functions for to the tools will be called. 
workflow.add_node("agent", run_tool_agent)
# Add tool invocation node to the graph
workflow.add_node("action", execute_tools)
# Define which node the graph will invoke at start.
workflow.set_entry_point("agent")

# Add flow logic with static edge.
# Each time a tool is invoked and completed we want to 
# return the result to the agent to assess if task is complete or to take further actions 

#each action invocation has an edge leading to the agent node.
workflow.add_edge('action', 'agent')


# Add flow logic with conditional edge.
workflow.add_conditional_edges(
    # first parameter is the starting node for the edge
    "agent",
    # the second parameter specifies the logic function to be run
    # to determine which node the edge will point to given the state.
    should_continue,

    #third parameter defines the mapping between the logic function 
    #output and the nodes on the graph
    # For each possible output of the logic function there must be a valid node.
    {
        # If 'continue' we proceed to the action node.
        "CONTINUE": "action",
        # Otherwise we end invocations with the END node.
        "END": END
    }
)

 

In [ ]:

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

# Finally, compile the graph! 
# This compiles it into a LangChain Runnable,
app = workflow.compile(checkpointer = memory)

 

In [ ]:

from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

 

In [ ]:

inputs = {"input": "hello", "chat_history": []} 
config = {"configurable": {"thread_id": "1"}}

for s in app.stream(inputs, config = config): 
    print(list(s.values())[0])
    print("----")

 

In [ ]:

memory.get(config)

 

In [ ]:

inputs_2 = {"input": "give me 1+1 and then 2 times 2", "chat_history": []} 
config_2 = {"configurable": {"thread_id": "2"}}


for s in app.stream(inputs_2, config = config_2): # 
    print(list(s.values())[0])
    print("----")

 

In [ ]:

memory.get(config_2)

 

 

3) Multi-Agent 구현하기¶

• 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

 

In [ ]:

import getpass
import os


# def _set_if_undefined(var: str):
#     if not os.environ.get(var):
#         os.environ[var] = getpass.getpass(f"Please provide your {var}")


# _set_if_undefined("OPENAI_API_KEY")
# _set_if_undefined("TAVILY_API_KEY")

 

In [ ]:

SECRET_ENV = os.getenv("TAVILY_API_KEY")
SECRET_ENV

 

In [ ]:

from typing import Annotated

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_experimental.utilities import PythonREPL

tavily_tool = TavilySearchResults(max_results=5)

# Warning: This executes code locally, which can be unsafe when not sandboxed

repl = PythonREPL()


@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """Use this to execute python code. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    try:
        result = repl.run(code)
    except BaseException as e:
        return f"Failed to execute. Error: {repr(e)}"
    result_str = f"Successfully executed:\n```python\n{code}\n```\nStdout: {result}"
    return (
        result_str + "\n\nIf you have completed all tasks, respond with FINAL ANSWER."
    )

 

In [ ]:

def make_system_prompt(suffix: str) -> str:
    return (
        "You are a helpful AI assistant, collaborating with other assistants."
        " Use the provided tools to progress towards answering the question."
        " If you are unable to fully answer, that's OK, another assistant with different tools "
        " will help where you left off. Execute what you can to make progress."
        " If you or any of the other assistants have the final answer or deliverable,"
        " prefix your response with FINAL ANSWER so the team knows to stop."
        f"\n{suffix}"
    )

 

In [ ]:

from typing import Literal

from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.prebuilt import create_react_agent
from langgraph.graph import MessagesState, END
from langgraph.types import Command
from langchain_openai import ChatOpenAI


llm = ChatOpenAI(model="gpt-4o-mini", api_key=os.environ['OPENAI_API_KEY'])

 

In [ ]:

def get_next_node(last_message: BaseMessage, goto: str):
    if "FINAL ANSWER" in last_message.content:
        # Any agent decided the work is done
        return END
    return goto


# Research agent and node
research_agent = create_react_agent(
    llm,
    tools=[tavily_tool],
    prompt=make_system_prompt(
        "You can only do research. You are working with a chart generator colleague."
    ),
)


def research_node(
    state: MessagesState,
) -> Command[Literal["chart_generator", END]]:
    result = research_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "chart_generator")
    # wrap in a human message, as not all providers allow
    # AI message at the last position of the input messages list
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="researcher"
    )
    return Command(
        update={
            # share internal message history of research agent with other agents
            "messages": result["messages"],
        },
        goto=goto,
    )


# Chart generator agent and node
# NOTE: THIS PERFORMS ARBITRARY CODE EXECUTION, WHICH CAN BE UNSAFE WHEN NOT SANDBOXED
chart_agent = create_react_agent(
    llm,
    [python_repl_tool],
    prompt=make_system_prompt(
        "You can only generate charts. You are working with a researcher colleague."
    ),
)


def chart_node(state: MessagesState) -> Command[Literal["researcher", END]]:
    result = chart_agent.invoke(state)
    goto = get_next_node(result["messages"][-1], "researcher")
    # wrap in a human message, as not all providers allow
    # AI message at the last position of the input messages list
    result["messages"][-1] = HumanMessage(
        content=result["messages"][-1].content, name="chart_generator"
    )
    return Command(
        update={
            # share internal message history of chart agent with other agents
            "messages": result["messages"],
        },
        goto=goto,
    )

 

In [ ]:

from langgraph.graph import StateGraph, START

workflow = StateGraph(MessagesState)
workflow.add_node("researcher", research_node)
workflow.add_node("chart_generator", chart_node)

workflow.add_edge(START, "researcher")
graph = workflow.compile()

 

In [ ]:

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # This requires some extra dependencies and is optional
    pass

 

In [ ]:

events = graph.stream(
    {
        "messages": [
            (
                "user",
                "First, get the UK's GDP over the past 5 years, then make a line chart of it. "
                "Once you make the chart, finish.",
            )
        ],
    },
    # Maximum number of steps to take in the graph
    {"recursion_limit": 150},
)
for s in events:
    print(s)
    print("----")

 

In [ ]:

from langchain_core.runnables.graph import MermaidDrawMethod
return Image(graph.get_graph().draw_mermaid_png(draw_method=MermaidDrawMethod.PYPPETEER))