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
        
        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()))

No description has been provided for this image

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))

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

1. LangChain 이해하기¶

1) 기본 개념¶

(1) LangChain Expression Language(LCEL)¶

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

(3) Runnable의 종류¶

(4) RunnableSequence : Runnable instance 묶기¶

(5) Runtime Arguments Binding¶

2) Messages¶

3) Agent란?¶

(1) Tool & Tool Binding¶

(2) Agent와 AgentExecutor¶

[1] 톺아보기¶

[2] 예제¶

(+) 개념 설명¶

[1]. Thought¶

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

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

✅ 비교 요약¶

[2]. Actions¶

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

구현 방안 2. Code Agent¶

[3]. Observations¶

2. LangGraph¶

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

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

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

(2) Sub-graph¶

+) Agent와 WorkFlow 구분하기¶

3. 실전 예제¶

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

2) Multi-Tool Agent 구현하기¶

3) Multi-Agent 구현하기¶