[langgraph] 도구(tool) 내에 이벤트를 스트리밍하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/streaming-events-from-within-tools/

그래프에 LLM을 호출하는 도구(또는 다른 LangChain Runnable 객체들, 예: 다른 그래프, LCEL 체인, 검색기 등)가 포함되어 있다면, 도구 실행 중에 부분 결과를 사용자에게 전달하고 싶을 수 있다. 특히 도구가 실행되는 데 시간이 오래 걸릴 경우 유용하다.

일반적인 시나리오는 LLM을 호출하는 도구가 생성한 LLM 토큰을 스트리밍하는 것이지만, 이는 Runnable 객체를 사용하는 모든 경우에 적용된다.

이 가이드는 도구 내에서 데이터를 스트리밍하는 방법을 보여준다. astream API와 stream_mode="messages"를 사용한 방법과 더 세분화된 astream_events API를 사용할 수 있다. 대부분의 경우 astream API로 충분하다.

준비

우선, 필요한 패키지를 설치하자.

pip install langgraph langchain-openai

그래프 정의

이 가이드에서는 간단한 ReAct 에이전트를 사용한다.

import asyncio

from dotenv import load_dotenv
from langchain_core.callbacks import Callbacks
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

load_dotenv()


@tool
async def get_items(
    place: str,
    callbacks: Callbacks,  # 직접 callbacks를 받는다 (Python <= 3.10인 경우는 필요하다)
) -> str:
    """Use this tool to look up which items are in the given place."""
    # async를 사용할 때는 LLM을 ainvoke를 사용하여 호출해야 한다!
    # 그렇지 않으면 스트리밍이 작동하지 않는다.
    return await llm.ainvoke(
        [
            {
                "role": "user",
                "content": f"다음 장소에서 어떤 물건을 찾을 수 있는지 알려줄래? : '{place}'. "
                "최소한 3개의 물건을 나열하고, 각 항목을 쉼표로 구분해줘. 그리고 각 물건에 대한 간단한 설명을 포함해.",
            }
        ],
        {"callbacks": callbacks},
    )


llm = ChatOpenAI(model_name="gpt-4o-mini")
tools = [get_items]
agent = create_react_agent(llm, tools=tools)

stream_mode="messages" 사용하기

stream_mode="messages"를 사용하는 것은 노드 내에 복잡한 LCEL 로직이 없거나 LCEL 체인 내에서 매우 세분화된 진행 상황이 필요하지 않은 경우 좋은 선택이다.

async def stream_content():
    final_message = ""
    async for msg, metadata in agent.astream(
        {"messages": [("human", "선반에 어떤 물건이 있어?")]},
        stream_mode="messages",
    ):
        # Stream all messages from the tool node
        if (
            msg.content
            and not isinstance(msg, HumanMessage)
            and metadata["langgraph_node"] == "tools"
            and not msg.name
        ):
            print(msg.content, end="|", flush=True)
        # Final message should come from our agent
        if msg.content and metadata["langgraph_node"] == "agent":
            final_message += msg.content


asyncio.run(stream_content())

stream event API 사용하기

단순화하기 위해, get_items 도구는 내부에 복잡한 LCEL 로직을 사용하지 않고, 단지 LLM을 호출한다.

하지만 도구가 더 복잡하다면 (예: 내부에 RAG 체인을 사용하고), 체인 내에서 더 세분화된 이벤트를 보고 싶다면, astream events API를 사용할 수 있다.

아래 예시는 API를 호출하는 방법만 보여준다.

from langchain_core.messages import HumanMessage

async def stream_content():
    async for event in agent.astream_events(
        {"messages": [{"role": "user", "content": "침실에 뭐가 있어?"}]}, version="v2"
    ):
        if (
            event["event"] == "on_chat_model_stream"
            and event["metadata"].get("langgraph_node") == "tools"
        ):
            print(event["data"]["chunk"].content, end="|", flush=True)

asyncio.run(stream_content())

|침|실|에서| 찾|을| 수| 있는| 물|건|은| 다음|과| 같습니다|:

|1|.| 침|대| -| 잠|을| 자|거나| 휴|식을| 취|하기| 위해| 사용하는| 가|구|로|,| 매|트|리스|와| 침|대| 프|레|임|으로| 구성|되어| 있습니다|.
|2|.| 옷|장| -| 의|류|와| 액|세|서|리를| 보|관|하는| 가|구|로|,| 서|랍|이나| 옷|걸|이가| 있어| 정|리|하기| 편|리|합니다|.
|3|.| 탁|상|등| -| 침|대| 옆|에| 놓|여|지는| 조|명| 기|구|로|,| 책|을| 읽|거나| 밤|에| 필요한| 조|명을| 제공합니다|.||

LangGraph 참고 자료

저작자표시

준비

그래프 정의

stream_mode="messages" 사용하기

stream event API 사용하기

LangGraph 참고 자료

티스토리툴바