LangChain 백엔드 응답을 스트리밍으로 바꾸기 — FastAPI에서 실시간 답변과 agent 진행 상황을 어떻게 흘려보낼까

LangChain 백엔드 응답을 스트리밍으로 바꾸기 — FastAPI에서 실시간 답변과 agent 진행 상황을 어떻게 흘려보낼까

여기서부터 UX가 확 달라집니다.

지금까지 만든 백엔드는 기본적으로 이런 방식이었죠.

1. 사용자가 질문을 보낸다
2. 서버가 내부에서 agent를 실행한다
3. 전부 끝난 뒤 최종 응답을 한 번에 돌려준다

이 방식은 구현이 단순하고 안정적입니다.
하지만 사용자는 중간에 아무것도 안 보이기 때문에, 조금만 느려져도 답답하게 느껴집니다. LangChain 문서는 streaming이 LLM 앱의 responsiveness를 높이는 핵심 기능이라고 설명하고, 전체 응답이 끝나기 전에 점진적으로 보여주면 UX가 크게 좋아진다고 말합니다. (LangChain Docs)

그리고 agent가 들어가면 더 그렇습니다.

• 지금 문서 검색 중인지
• tool을 호출했는지
• 이미 답변 생성은 시작됐는지

이런 걸 보여주면 사용자는 “멈춘 것 같은” 느낌을 덜 받습니다. LangGraph streaming 문서도 스트리밍의 목적을 실시간 업데이트를 surface하는 데 둔다고 설명합니다. (LangChain Docs)

---

왜 스트리밍이 백엔드에서 중요할까

스트리밍은 단순히 보기 좋게 만드는 기능이 아닙니다.
백엔드 구조 자체를 더 “대화형”으로 느끼게 만드는 장치예요.

FastAPI 공식 문서는 StreamingResponse가 데이터를 조각 단위로 흘려보낼 수 있게 해주며, FastAPI는 각 chunk를 그대로 전달하고 JSON 변환을 시도하지 않는다고 설명합니다. 또 긴 스트림이나 무한 스트림에 특히 중요하다고 안내합니다. (FastAPI)

즉, 스트리밍을 붙이면 이런 게 가능해집니다.

• 답변 토큰을 순서대로 보여주기
• “tool 호출 시작” 같은 이벤트를 중간에 알려주기
• 최종 structured result가 나오기 전까지 진행 상황 보여주기

이건 특히 agent에 잘 맞습니다.
agent는 원래 내부적으로 여러 step을 거치기 때문입니다. LangChain agents 문서는 agent가 tool loop를 돌며 목표를 해결한다고 설명합니다. (LangChain Docs)

---

오늘 글에서 만들 것

이번 글에서는 가장 현실적인 시작점으로 갑니다.

• FastAPI에서 StreamingResponse 사용
• newline-delimited JSON, 즉 JSON Lines 비슷한 방식으로 chunk 전송
• LangChain/LangGraph streaming 이벤트를 받아서 흘려보내기
• 마지막에는 structured payload 형태의 final 이벤트 보내기

여기서 일부러 SSE 전용 라이브러리까지는 가지 않겠습니다.
FastAPI 자체만으로도 충분히 시작할 수 있기 때문입니다. FastAPI는 StreamingResponse를 공식적으로 지원합니다. (FastAPI)

---

왜 JSON Lines 스타일이 좋은가

스트리밍을 처음 붙일 때 흔히 두 가지가 나옵니다.

• SSE
• chunked plain text
• JSON Lines

이번 글에서는 JSON Lines 스타일로 가겠습니다.
이유는 단순합니다.

• 백엔드 구현이 간단함
• 프론트에서 줄 단위로 파싱하기 쉬움
• 텍스트 토큰과 상태 이벤트를 같은 채널로 다루기 편함

FastAPI 문서도 스트림 데이터는 chunk를 그대로 전달한다고 설명하고, raw Response를 직접 반환하면 자동 검증과 문서화가 사라진다고 말합니다. 즉, 스트리밍에서는 “문서화되는 response_model”보다 실제 전송 포맷을 명확히 설계하는 쪽이 중요해집니다. (FastAPI)

---

오늘 만들 스트리밍 이벤트 형태

이번 글에서는 이런 식으로 줄 단위 JSON을 흘려보내겠습니다.

{"type":"status","message":"agent_started"}
{"type":"token","content":"재택근무는 "}
{"type":"token","content":"팀 운영 원칙에 따라 "}
{"type":"token","content":"주 2회까지 가능합니다."}
{"type":"final","data":{"answer":"재택근무는 팀 운영 원칙에 따라 주 2회까지 가능합니다.","response_type":"policy_answer","used_tools":["search_policy_documents"],"sources":["policy-remote-work"]}}

이 구조가 좋은 이유는 분명합니다.

• status: 진행 상황
• token: 실시간 텍스트
• final: 최종 structured payload

즉, 프론트는 토큰은 화면에 바로 붙이고, 마지막 final은 상태 저장이나 후속 UI 제어에 쓸 수 있습니다.

---

프로젝트 구조

지난 글 구조를 그대로 이어갑니다.

project/
├─ app/
│  ├─ main.py
│  ├─ schemas.py
│  ├─ agent.py
│  └─ tools.py
├─ requirements.txt
└─ .env

이번 글에서 바뀌는 핵심은 main.py와 agent.py입니다.

---

requirements.txt

이번 글은 LangChain/LangGraph streaming과 FastAPI StreamingResponse를 같이 쓰는 흐름입니다.

fastapi==0.112.0
uvicorn[standard]==0.30.6
langchain==1.0.3
langchain-openai==1.0.1
langgraph==1.0.2
pydantic==2.11.7
python-dotenv==1.0.1

FastAPI는 실제 앱에서 버전 pinning을 권장하고, LangChain v1은 create_agent를 표준 방식으로 설명합니다. (LangChain Docs)

---

1. 스키마는 일반 응답과 스트리밍 요청을 같이 쓸 수 있게 유지하기

app/schemas.py

from typing import Literal
from pydantic import BaseModel, Field


class AskRequest(BaseModel):
    session_id: str = Field(description="대화 세션 ID. LangChain thread_id로 사용됨")
    question: str = Field(description="사용자 질문")


class AssistantStructuredResponse(BaseModel):
    answer: str = Field(description="사용자에게 보여줄 최종 답변")
    response_type: Literal["policy_answer", "request_status", "summary", "unknown"] = Field(
        description="응답 유형"
    )
    used_tools: list[str] = Field(
        default_factory=list,
        description="에이전트가 사용한 tool 이름 목록"
    )
    sources: list[str] = Field(
        default_factory=list,
        description="참고한 출처 또는 문서 소스 목록"
    )


class AskResponse(BaseModel):
    answer: str = Field(description="사용자에게 보여줄 최종 답변")
    response_type: Literal["policy_answer", "request_status", "summary", "unknown"] = Field(
        description="응답 유형"
    )
    used_tools: list[str] = Field(default_factory=list)
    sources: list[str] = Field(default_factory=list)

FastAPI는 request body와 response model에 Pydantic 모델을 쓰는 방식을 기본으로 설명합니다. (FastAPI)

여기서 포인트는,
일반 JSON 응답용 AskResponse는 그대로 유지하고,
스트리밍은 별도 endpoint로 추가한다는 점입니다.

---

2. tools.py는 그대로 둬도 된다

이번 글은 스트리밍 UX가 핵심이므로 tool은 지난 글과 동일하게 둡니다.

app/tools.py

import json
from langchain.tools import tool


@tool
def search_policy_documents(query: str) -> str:
    """Search internal policy documents by query.

    Use this tool when the user asks about company policy, rules, benefits,
    leave, expense, onboarding, or internal process documentation.
    """
    docs = [
        {
            "title": "연차 사용 정책",
            "content": "정규직 직원은 연차를 사내 인사 시스템에서 신청할 수 있으며, 팀 리더 승인 후 확정된다.",
            "source": "policy-annual-leave",
        },
        {
            "title": "출장비 정산 정책",
            "content": "출장비는 결제 영수증과 함께 경비 시스템에 등록해야 하며, 귀사 후 7일 이내 정산이 원칙이다.",
            "source": "policy-expense-travel",
        },
        {
            "title": "재택근무 정책",
            "content": "재택근무는 팀 운영 원칙에 따라 주 2회까지 가능하며, 예외는 팀장 승인 후 가능하다.",
            "source": "policy-remote-work",
        },
    ]

    query_lower = query.lower()
    matched = []

    for doc in docs:
        haystack = f"{doc['title']} {doc['content']}".lower()
        if any(keyword in haystack for keyword in query_lower.split()):
            matched.append(doc)

    if not matched:
        return json.dumps(
            {
                "found": False,
                "results": [],
                "message": "관련 정책 문서를 찾지 못했습니다.",
            },
            ensure_ascii=False,
        )

    return json.dumps(
        {
            "found": True,
            "results": matched[:2],
        },
        ensure_ascii=False,
    )


@tool
def get_request_status(request_id: str) -> str:
    """Get the status of an internal request or ticket by request ID.

    Use this tool when the user asks for the latest state of a request,
    ticket, approval, or internal work item.
    """
    mock_requests = {
        "1024": {
            "request_id": "1024",
            "status": "승인 대기",
            "owner": "총무팀",
            "summary": "노트북 교체 요청",
        },
        "2048": {
            "request_id": "2048",
            "status": "처리 완료",
            "owner": "IT 지원팀",
            "summary": "사내 VPN 권한 부여",
        },
        "3001": {
            "request_id": "3001",
            "status": "진행 중",
            "owner": "재무팀",
            "summary": "출장비 정산 검토",
        },
    }

    data = mock_requests.get(request_id)
    if not data:
        return json.dumps(
            {
                "found": False,
                "request_id": request_id,
                "message": "해당 요청 ID를 찾지 못했습니다.",
            },
            ensure_ascii=False,
        )

    return json.dumps(
        {
            "found": True,
            "request": data,
        },
        ensure_ascii=False,
    )


@tool
def summarize_for_employee(text: str) -> str:
    """Summarize internal information for an employee in short, clear Korean.

    Use this tool when the user explicitly asks for a short summary
    or when multiple pieces of information need to be condensed.
    """
    short = text.strip().replace("\n", " ")
    if len(short) > 180:
        short = short[:180] + "..."
    return f"요약: {short}"

LangChain tools는 well-defined inputs and outputs를 가진 callable function으로 agents에 연결된다고 설명됩니다. (LangChain Docs)

---

3. agent.py에 일반 응답과 스트리밍 응답을 둘 다 지원하게 만들기

반응형

이번 글의 핵심은 여기입니다.

LangChain streaming 문서는 스트리밍을 세 가지 mode로 설명합니다.

• updates: agent progress
• messages: LLM tokens와 metadata
• custom: 사용자 정의 이벤트 (LangChain Docs)

우리는 여기서 messages와 updates 감각을 같이 활용하되, 가장 단순한 코드로 JSON Lines를 만들겠습니다.

app/agent.py

import json
import os
from functools import lru_cache
from typing import AsyncIterator

from dotenv import load_dotenv
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

from .schemas import AssistantStructuredResponse
from .tools import (
    get_request_status,
    search_policy_documents,
    summarize_for_employee,
)

load_dotenv()


@lru_cache(maxsize=1)
def get_agent():
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("OPENAI_API_KEY 환경변수가 설정되어 있지 않습니다.")

    checkpointer = InMemorySaver()

    agent = create_agent(
        model="openai:gpt-4o-mini",
        tools=[
            search_policy_documents,
            get_request_status,
            summarize_for_employee,
        ],
        system_prompt=(
            "너는 사내 업무 도우미다. "
            "정책/규정/제도 질문은 정책 검색 tool을 사용해라. "
            "요청/티켓/상태 확인 질문은 요청 상태 조회 tool을 사용해라. "
            "사용자가 요약을 원하거나 정보가 길면 요약 tool을 사용해라. "
            "같은 thread 안에서는 이전 대화 문맥을 참고해라. "
            "문서나 tool 결과에 없는 내용은 추측하지 마라. "
            "최종 응답은 반드시 주어진 structured response schema에 맞춰라."
        ),
        checkpointer=checkpointer,
        response_format=AssistantStructuredResponse,
    )
    return agent


def ask_assistant(session_id: str, question: str) -> AssistantStructuredResponse:
    agent = get_agent()

    result = agent.invoke(
        {
            "messages": [{"role": "user", "content": question}]
        },
        {
            "configurable": {
                "thread_id": session_id,
            }
        },
    )

    return result["structured_response"]


async def stream_assistant(session_id: str, question: str) -> AsyncIterator[str]:
    agent = get_agent()

    yield json.dumps(
        {"type": "status", "message": "agent_started"},
        ensure_ascii=False,
    ) + "\n"

    final_text_parts: list[str] = []
    used_tools: list[str] = []

    async for chunk, metadata in agent.astream(
        {
            "messages": [{"role": "user", "content": question}]
        },
        {
            "configurable": {
                "thread_id": session_id,
            }
        },
        stream_mode="messages",
    ):
        # chunk가 AIMessageChunk류인 경우 token/content를 모아서 보냄
        text = ""
        if hasattr(chunk, "content"):
            if isinstance(chunk.content, str):
                text = chunk.content
            elif isinstance(chunk.content, list):
                text = "".join(
                    item.get("text", "")
                    for item in chunk.content
                    if isinstance(item, dict)
                )

        if text:
            final_text_parts.append(text)
            yield json.dumps(
                {"type": "token", "content": text},
                ensure_ascii=False,
            ) + "\n"

        langgraph_node = metadata.get("langgraph_node")
        if langgraph_node == "tools":
            tool_name = metadata.get("name")
            if tool_name and tool_name not in used_tools:
                used_tools.append(tool_name)
                yield json.dumps(
                    {"type": "tool", "name": tool_name},
                    ensure_ascii=False,
                ) + "\n"

    final_answer = "".join(final_text_parts).strip()

    # 마지막엔 invoke로 structured_response를 한 번 더 안정적으로 받는다.
    structured = ask_assistant(session_id=session_id, question=question)

    yield json.dumps(
        {
            "type": "final",
            "data": {
                "answer": structured.answer or final_answer,
                "response_type": structured.response_type,
                "used_tools": structured.used_tools or used_tools,
                "sources": structured.sources,
            },
        },
        ensure_ascii=False,
    ) + "\n"

여기서 핵심은 두 가지입니다.

첫째, agent.astream(..., stream_mode="messages")를 사용했다는 점입니다. LangChain streaming 문서는 stream_mode="messages"가 LLM token과 metadata를 실시간으로 주는 방식이라고 설명합니다. (LangChain Docs)

둘째, 마지막에는 structured_response를 얻기 위해 ask_assistant()를 다시 한 번 호출했다는 점입니다.

이건 학습용으로 가장 이해하기 쉬운 방식입니다.
다만 실제 운영에서는 추가 호출 비용이 생기므로, 나중에는 updates 모드와 structured output 흐름을 더 세밀하게 맞추거나 LangGraph 노드 단계에서 직접 캡처하는 방식으로 최적화하는 것이 좋습니다. LangChain v1 문서는 structured output이 메인 루프 안에 통합되어 있고, create_agent가 production-ready runtime을 제공한다고 설명합니다. (LangChain Docs)

---

4. FastAPI에서 StreamingResponse 엔드포인트 추가하기

app/main.py

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse

from .agent import ask_assistant, stream_assistant
from .schemas import AskRequest, AskResponse

app = FastAPI(
    title="Internal Assistant API",
    version="0.4.0",
    description="LangChain streaming + structured output이 붙은 사내 도우미 백엔드 예제",
)


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/assistant/ask", response_model=AskResponse)
def ask(request: AskRequest) -> AskResponse:
    try:
        structured = ask_assistant(
            session_id=request.session_id,
            question=request.question,
        )
        return AskResponse(
            answer=structured.answer,
            response_type=structured.response_type,
            used_tools=structured.used_tools,
            sources=structured.sources,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/assistant/stream")
async def stream(request: AskRequest):
    return StreamingResponse(
        stream_assistant(
            session_id=request.session_id,
            question=request.question,
        ),
        media_type="application/x-ndjson",
    )

FastAPI의 StreamingResponse는 generator나 async generator를 받아 chunk를 그대로 전달합니다. 스트림 데이터 문서와 custom response 문서 모두 이 흐름을 안내합니다. 또한 raw Response를 직접 반환하면 자동 검증/문서화가 사라지기 때문에, 일반 JSON endpoint와 스트리밍 endpoint를 분리하는 방식이 자연스럽습니다. (FastAPI)

media_type을 application/x-ndjson로 둔 이유는
newline-delimited JSON 스트림이라는 의도를 분명히 하기 위해서입니다.

---

5. 실행하고 스트리밍 테스트해보기

uvicorn app.main:app --reload

테스트:

curl -N -X POST "http://127.0.0.1:8000/assistant/stream" \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "ops-stream-1",
    "question": "재택근무 정책 알려주고 한 줄로 정리해줘."
  }'

-N은 curl 버퍼링을 끄는 옵션입니다.
이렇게 하면 줄 단위 이벤트가 도착하는 대로 보입니다.

예상 출력은 대략 이런 느낌입니다.

{"type":"status","message":"agent_started"}
{"type":"tool","name":"search_policy_documents"}
{"type":"token","content":"재택근무는 "}
{"type":"token","content":"팀 운영 원칙에 따라 "}
{"type":"token","content":"주 2회까지 가능하며, "}
{"type":"token","content":"예외는 팀장 승인 후 가능합니다."}
{"type":"final","data":{"answer":"재택근무는 팀 운영 원칙에 따라 주 2회까지 가능하며, 예외는 팀장 승인 후 가능합니다.","response_type":"summary","used_tools":["search_policy_documents","summarize_for_employee"],"sources":["policy-remote-work"]}}

LangChain streaming은 토큰과 진행 상태를 실시간으로 surface하는 데 목적이 있다고 설명합니다. (LangChain Docs)

---

6. 왜 일반 endpoint와 스트리밍 endpoint를 둘 다 두는 게 좋을까

이건 꽤 실전적인 선택입니다.

일반 /assistant/ask

• 구현 단순
• 테스트 쉬움
• 프론트가 간단할 때 좋음
• 배치/백오피스/서버 간 호출에 좋음

스트리밍 /assistant/stream

• UX 좋음
• agent 진행감을 보여주기 좋음
• 답변이 길거나 tool이 많을 때 좋음

FastAPI는 일반 JSON response model과 StreamingResponse를 모두 자연스럽게 지원합니다. response model은 자동 검증과 문서화에 강하고, streaming은 chunk 전송에 강합니다. (FastAPI)

즉, 둘 중 하나만 택하는 게 아니라
용도별로 둘 다 두는 게 실제로 더 낫습니다.

---

7. 지금 방식의 한계도 분명하다

좋은 점만 보면 안 됩니다.

한계 1. final을 위해 한 번 더 invoke한다

지금 예제는 이해를 쉽게 하려고 그렇게 했습니다.
하지만 그만큼 추가 비용과 지연이 생깁니다. LangChain structured output은 원래 메인 루프 안에 통합돼 있으므로, 운영에서는 이를 더 직접적으로 활용하는 편이 낫습니다. (LangChain Docs)

한계 2. token stream과 structured result가 완전히 같은 값이 아닐 수 있다

스트리밍으로 나온 텍스트와 마지막 structured result가 미세하게 다를 수 있습니다.

한계 3. tool 이벤트 추적이 단순하다

지금은 metadata에서 langgraph_node == "tools" 정도만 보는 아주 단순한 방식입니다.
운영에선 더 풍부한 observability가 필요합니다.

한계 4. 프론트가 ndjson 파싱을 구현해야 한다

JSON 한 번 받는 것보다 조금 더 손이 갑니다.

그래도 시작점으로는 충분히 좋습니다.
왜냐면 핵심 감각이 다 보이거든요.

• status
• token
• final

---

8. 그럼 나중엔 어떻게 더 좋아지나

여기서부터는 확장 방향입니다.

1) updates 모드도 함께 사용

LangChain streaming 문서는 stream_mode="updates"가 agent progress를 step 단위로 보여준다고 설명합니다. 그러면 tool 시작/종료를 더 명확히 표현할 수 있습니다. (LangChain Docs)

2) structured output을 한 번에 끝내는 방식으로 최적화

현재 예제는 이해용이라 마지막에 한 번 더 호출했습니다.
운영에선 agent graph state나 middleware 단계에서 최종 structured payload를 더 직접적으로 캡처하는 편이 낫습니다. LangChain v1이 middleware와 graph runtime을 강조하는 이유도 이런 제어 확장성 때문입니다. (LangChain Docs)

3) SSE로 바꾸기

웹 프론트엔드에서는 SSE가 더 익숙할 수 있습니다.
다만 오늘은 FastAPI 기본 스트리밍만으로도 충분히 시작 가능해서 ndjson으로 갔습니다. FastAPI는 StreamingResponse 기반 스트림을 공식 지원합니다. (FastAPI)

4) streaming + short-term memory + persistence 같이 묶기

지금 구조는 이미 thread_id와 checkpointer를 쓰고 있으니, 나중에 persistent saver로 교체해도 흐름이 이어집니다. short-term memory는 thread-scoped state로 저장되며 checkpointer를 통해 재개 가능하다고 LangChain이 설명합니다. (LangChain Docs)

---

9. 스트리밍에서 취소 처리를 왜 생각해야 하나

이건 꽤 운영스러운 이야기입니다.

사용자가 화면을 닫거나 요청을 취소할 수 있죠.
FastAPI 스트림 문서는 cancellation을 고려해야 하고, 긴 스트림에서는 cooperative cancellation이 특히 중요하다고 설명합니다. (FastAPI)

즉, 실제 운영에선 이런 걸 고려해야 합니다.

• 클라이언트 연결 종료 시 generator 종료
• tool 호출 중이면 timeout/취소 처리
• 너무 긴 스트림은 적절히 마감

오늘 예제는 단순 버전이라 여기까진 안 갔지만,
스트리밍은 일반 JSON endpoint보다 이런 운영 포인트가 더 중요합니다.

---

10. 이 단계에서 프론트는 어떻게 붙이면 좋나

가장 단순한 방식은 이겁니다.

• fetch로 POST 호출
• response.body reader로 chunk 읽기
• 줄바꿈 기준으로 ndjson 파싱
• type에 따라 UI 업데이트

즉:

• status → 로딩 문구
• tool → “정책 검색 중…” 같은 배지
• token → 답변 타이핑 효과
• final → 상태 저장, 출처 렌더링, 후속 버튼 노출

이게 가능한 이유는, 우리가 응답 계약을 문자열이 아니라 이벤트 스트림 계약으로 바꿨기 때문입니다.

---

오늘 글에서 꼭 가져가야 할 한 문장

오늘 내용을 한 줄로 줄이면 이겁니다.

LangChain streaming과 FastAPI StreamingResponse를 붙이면, 백엔드는 최종 답 하나를 기다리는 서버가 아니라 진행 상태와 토큰을 실시간으로 흘려보내는 대화형 인터페이스가 된다.

이 감각이 생기면
이제 “AI 백엔드”가 훨씬 살아 있는 시스템처럼 보이기 시작합니다.

---

마무리

저는 스트리밍을 처음 붙였을 때
“성능이 빨라진 건 아닌데, 왜 이렇게 훨씬 좋아 보이지?”
이 생각을 했습니다.

답은 단순했습니다.

사용자는 기다리는 시간을 싫어하는 게 아니라,
아무 일도 안 일어나는 것처럼 보이는 시간을 싫어하더라고요.

스트리밍은 그 공백을 메워줍니다.

• 지금 시작됐고
• 지금 검색 중이고
• 지금 답변을 만들고 있고
• 거의 끝나간다

이걸 보여주는 것만으로도 체감이 꽤 달라집니다.

그래서 저는 스트리밍이 단순한 장식이 아니라,
대화형 AI 서비스의 핵심 UX 중 하나라고 생각합니다.

---

다음 글 예고

다음 글에서는
LangChain 백엔드 운영 준비하기 — 로그, 관측, 디버깅을 어디서부터 붙여야 하나
로 이어가겠습니다.

이제부터는 “기능이 된다”를 넘어서,
문제가 생겼을 때 어디서 막혔는지 보고 고칠 수 있는 운영 감각으로 넘어가게 될 거예요.

출처

• LangChain Streaming 문서: streaming은 실시간 업데이트를 surface해 responsiveness를 높이고 UX를 개선하며, updates, messages, custom 같은 mode를 제공한다고 설명합니다. (LangChain Docs)
• LangChain Agents 문서: create_agent는 production-ready agent implementation이며 tool loop를 실행한다고 설명합니다. (LangChain Docs)
• LangChain v1 문서: create_agent는 LangChain 1.0의 standard way to build agents이며 middleware 중심 확장을 지원합니다. (LangChain Docs)
• LangChain Structured Output 문서: create_agent(..., response_format=...)는 structured output을 자동 처리하고 결과를 structured_response에 담는다고 설명합니다. (LangChain Docs)
• LangChain Short-term Memory 및 Memory Overview 문서: short-term memory는 thread-scoped state이고, checkpointer를 통해 저장·재개된다고 설명합니다. (LangChain Docs)
• FastAPI Stream Data / Custom Response 문서: StreamingResponse는 chunk를 그대로 스트리밍하며, 긴 스트림에서는 cancellation handling이 중요하다고 설명합니다. (FastAPI)
• FastAPI Response Model 문서: response_model은 반환 데이터의 검증, 필터링, 문서화를 담당한다고 설명합니다. (FastAPI)
• FastAPI Return a Response Directly 문서: raw Response를 직접 반환하면 자동 검증·변환·문서화가 적용되지 않는다고 설명합니다. (FastAPI)

LangChain, FastAPI, StreamingResponse, LangGraph Streaming, AI Agent, 실시간스트리밍, 생성형AI, 백엔드개발, LangChain Backend, 주니어개발자