스트리밍

from langchain_core.messages import HumanMessage

for chunk in app.stream(
    {"messages": [HumanMessage(content="RAG란 무엇인가요?")]},
    config=config,
    stream_mode="values",
):
    # 매 노드 실행 후 전체 messages 상태
    messages = chunk["messages"]
    print(f"메시지 수: {len(messages)}")
    print(messages[-1].content[:100])

for chunk in app.stream(
    {"messages": [HumanMessage(content="질문")]},
    config=config,
    stream_mode="updates",
):
    # {노드이름: {변경된 상태}}
    for node_name, update in chunk.items():
        print(f"노드: {node_name}")
        if "messages" in update:
            print(f"새 메시지: {update['messages'][-1].content[:100]}")

for msg, metadata in app.stream(
    {"messages": [HumanMessage(content="LangGraph를 설명해줘")]},
    config=config,
    stream_mode="messages",
):
    # msg: 토큰 단위의 메시지 청크
    # metadata: 노드 이름, 실행 정보 등
    if msg.content:
        print(msg.content, end="", flush=True)

from langgraph.types import StreamWriter

def research_node(state: State, writer: StreamWriter) -> State:
    """검색 진행률을 스트리밍"""
    queries = ["쿼리1", "쿼리2", "쿼리3"]

    for i, query in enumerate(queries):
        results = retriever.invoke(query)
        # 커스텀 이벤트 전송
        writer({"progress": f"검색 {i+1}/{len(queries)}: {query}"})

    return {"messages": [("assistant", "검색 완료")]}

# custom 이벤트 수신
for chunk in app.stream(
    {"messages": [HumanMessage(content="조사해줘")]},
    config=config,
    stream_mode="custom",
):
    print(chunk)  # {"progress": "검색 1/3: 쿼리1"}

for event in app.stream(
    {"messages": [HumanMessage(content="질문")]},
    config=config,
    stream_mode=["messages", "updates"],
):
    mode = event[0]  # "messages" 또는 "updates"
    data = event[1]

    if mode == "messages":
        msg, metadata = data
        print(msg.content, end="")
    elif mode == "updates":
        print(f"\n[노드 업데이트: {list(data.keys())}]")

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

fastapi_app = FastAPI()

@fastapi_app.post("/chat")
async def chat(question: str):
    async def event_stream():
        async for msg, metadata in app.astream(
            {"messages": [HumanMessage(content=question)]},
            config=config,
            stream_mode="messages",
        ):
            if msg.content:
                yield f"data: {msg.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")