FastAPI로 LangChain 백엔드 만들기 — 사내 도우미를 API 서버 형태로 감싸는 가장 현실적인 시작

FastAPI로 LangChain 백엔드 만들기 — 사내 도우미를 API 서버 형태로 감싸는 가장 현실적인 시작

여기서부터는 느낌이 좀 달라집니다.

지금까지는 LangChain 예제를 터미널에서 실행해보는 흐름이 많았죠.
질문을 넣고, tool을 붙이고, agent를 돌리고, 결과를 출력해봤습니다.

근데 실제 서비스는 여기서 끝나지 않아요.
결국 누군가는 이 기능을 웹, 앱, 사내 시스템에서 써야 하거든요.
그러려면 LangChain 코드를 API 서버 형태로 감싸야 합니다.

그리고 Python 진영에서 이 시작점으로 제일 현실적인 선택 중 하나가 FastAPI입니다.
FastAPI 공식 문서는 FastAPI를 표준 Python type hints 기반의 modern, high-performance API framework로 소개하고, 요청 본문은 보통 Pydantic 모델로 선언한다고 설명합니다. 또 LangChain 쪽은 v1 기준 에이전트의 표준 생성 방식으로 create_agent를 안내하고, short-term memory는 agent state의 일부로 관리되며 checkpointer를 통해 thread 단위로 이어갈 수 있다고 설명해요. (FastAPI)

저는 이 구간이 되게 중요하다고 생각합니다.
왜냐면 여기서부터 코드가 “예제”에서 “서비스 구조”로 바뀌기 시작하거든요.

---

왜 FastAPI가 좋은 시작점일까

솔직히 사내 도우미 같은 건 꼭 FastAPI여야 하는 건 아닙니다.
Node, Spring, Django, Flask 다 가능하죠.

근데 LangChain Python을 쓰는 흐름이라면 FastAPI가 꽤 자연스럽습니다.

이유는 몇 가지가 있어요.

• Pydantic 기반이라 요청/응답 스키마를 잡기 편함
• JSON API 만들기가 단순함
• 의존성 주입 구조가 꽤 깔끔함
• 나중에 스트리밍이나 인증, 미들웨어 붙이기 좋음

FastAPI 공식 문서도 요청 본문은 Pydantic 모델로 선언하고, 응답도 response model을 주면 Pydantic 기반으로 직렬화한다고 설명합니다. 또 dependency system이 강력하면서도 단순하게 설계돼 있어서 컴포넌트를 주입하기 좋다고 안내해요. (FastAPI)

그러니까 지금 단계에선 이렇게 이해하면 됩니다.

LangChain이 AI 기능을 담당한다면, FastAPI는 그 기능을 다른 시스템이 쓸 수 있게 노출하는 껍데기다.

---

이번 글에서 만들 것

이번 글에서는 너무 거대한 구조로 가지 않겠습니다.
대신 진짜 많이 쓰는 최소 구조를 만들 거예요.

우리가 만들 API는 이런 식입니다.

• /health
  서버 상태 확인
• /assistant/ask
  질문을 보내면 사내 도우미가 답변 반환

그리고 이 사내 도우미는 지난 글 흐름을 그대로 이어갑니다.

• 정책 질문이면 정책 검색 tool
• 요청 상태 질문이면 상태 조회 tool
• 길면 요약

즉, 오늘 목표는 이겁니다.

LangChain agent를 FastAPI 안에서 호출 가능한 서비스로 감싸기

---

먼저 구조부터 잡아보자

작게 시작해도 폴더 구조는 너무 막 두지 않는 게 좋습니다.
처음엔 대충 한 파일로 해도 되지만, 금방 지저분해져요.

저는 처음엔 이 정도면 충분하다고 생각합니다.

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

역할은 간단합니다.

• main.py: FastAPI 앱과 라우트
• schemas.py: 요청/응답 Pydantic 모델
• agent.py: LangChain agent 생성 및 호출
• tools.py: 사내 도우미용 tool 모음

이런 분리는 FastAPI dependency 구조와도 잘 맞고, LangChain agent/tool 코드를 나중에 재사용하기도 편합니다. FastAPI 문서의 dependency system 설명과 LangChain의 tool/agent 분리 철학을 같이 생각하면 꽤 자연스러운 구조예요. (FastAPI)

---

requirements.txt

요즘 기준으로 너무 넓게 버전 범위를 풀어두면 나중에 예제가 갑자기 깨질 수 있습니다.
FastAPI 공식 문서도 실제 애플리케이션에서는 버전 pinning을 권장합니다. (FastAPI)

예시는 이렇게 잡겠습니다.

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

버전은 프로젝트 시점에 맞춰 조정할 수 있지만,
핵심은 FastAPI와 LangChain 계열을 명시적으로 고정해두는 겁니다. FastAPI는 버전 pinning을 권장하고, LangChain v1 문서는 create_agent가 1.0 기준 표준 방식이라고 설명합니다. (FastAPI)

---

1. 요청/응답 스키마부터 만들기

FastAPI에서 제일 좋은 습관 중 하나는
API 스키마를 먼저 잡는 겁니다.

app/schemas.py

from pydantic import BaseModel, Field


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


class AskResponse(BaseModel):
    answer: str = Field(description="사내 도우미의 최종 답변")

FastAPI 공식 문서는 요청 본문을 Pydantic 모델로 선언하는 방식을 기본으로 설명하고, response model을 선언하면 JSON 직렬화와 문서화에 도움이 된다고 안내합니다. (FastAPI)

이 단계가 중요한 이유는 단순합니다.

나중에 프론트엔드나 다른 시스템이 붙을 때
“이 API는 뭘 보내고 뭘 받는지”가 명확해지거든요.

---

2. 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 문서는 tool이 well-defined inputs/outputs를 가진 callable function이고, 이름·설명·인자 정보가 모델의 tool 사용 추론에 직접 영향을 준다고 설명합니다. 그래서 지금처럼 docstring을 꽤 분명하게 적어두는 게 중요해요. (LangChain Docs)

여기서 중요한 건 두 가지입니다.

첫째, 조회성 tool 중심으로 시작했다는 점.
둘째, 반환 형식을 최대한 일정하게 만들었다는 점.

이건 나중에 운영 단계에서 꽤 큰 차이를 만듭니다.

---

3. agent 생성 로직 분리하기

app/agent.py

import os
from functools import lru_cache

from dotenv import load_dotenv
from langchain.agents import create_agent

from .tools import (
    search_policy_documents,
    get_request_status,
    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 환경변수가 설정되어 있지 않습니다.")

    agent = create_agent(
        model="openai:gpt-4o-mini",
        tools=[search_policy_documents, get_request_status, summarize_for_employee],
        system_prompt=(
            "너는 사내 업무 도우미다. "
            "정책/규정/제도 질문은 정책 검색 tool을 사용해라. "
            "요청/티켓/상태 확인 질문은 요청 상태 조회 tool을 사용해라. "
            "사용자가 요약을 원하거나 정보가 길면 요약 tool을 사용해라. "
            "문서나 tool 결과에 없는 내용은 추측하지 마라."
        ),
    )
    return agent


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

    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]}
    )
    return result["messages"][-1].content

LangChain v1 문서는 create_agent가 standard way to build agents라고 설명하고, agent는 LangGraph 기반 runtime 위에서 동작합니다. 그러니까 이렇게 별도 모듈로 분리해두면 나중에 middleware, short-term memory, structured output을 붙이기도 좋습니다. short-term memory 문서도 agent state와 thread separation 개념을 설명하고요. (LangChain Docs)

여기서 @lru_cache를 쓴 이유는 매 요청마다 agent를 새로 조립하지 않기 위해서입니다.
아주 큰 최적화는 아니어도, 실제 서버 형태로 가면 이런 분리가 꽤 중요해집니다.

---

4. FastAPI 앱 만들기

반응형

app/main.py

from fastapi import FastAPI, HTTPException

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

app = FastAPI(
    title="Internal Assistant API",
    version="0.1.0",
    description="LangChain 기반 사내 도우미 백엔드 예제",
)


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


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

FastAPI 공식 문서는 요청 본문을 Pydantic 모델로 받고, response model을 선언하면 JSON 직렬화와 스키마 문서화가 자동으로 따라온다고 설명합니다. 또 기본적으로 JSON 응답을 반환하며, response model을 쓰는 편이 성능과 일관성 면에서 유리하다고 안내합니다. (FastAPI)

이 구조가 좋은 이유는 되게 현실적입니다.

• /health로 배포 후 상태 확인 가능
• /assistant/ask 하나만으로 프론트엔드와 연결 가능
• 요청/응답 스키마가 명확
• 내부 LangChain 로직은 FastAPI 밖으로 분리

이게 바로 “예제 코드”에서 “백엔드 API”로 넘어가는 첫걸음이에요.

---

5. 실행하기

루트에서 이렇게 실행합니다.

uvicorn app.main:app --reload

FastAPI 튜토리얼은 보통 Uvicorn으로 앱을 실행하는 흐름을 안내하고, FastAPI는 step by step tutorial 구조로 API 기능을 확장해가도록 설계돼 있습니다. (FastAPI)

실행 후에는:

• GET /health
• POST /assistant/ask

를 테스트해보면 됩니다.

예시 요청:

curl -X POST "http://127.0.0.1:8000/assistant/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "demo-session",
    "question": "재택근무 정책 알려줘."
  }'

예시 응답:

{
  "answer": "재택근무는 팀 운영 원칙에 따라 주 2회까지 가능하며, 예외는 팀장 승인 후 가능합니다."
}

---

6. 그런데 왜 session_id를 아직 안 썼을까

좋은 질문입니다.

요청 스키마에는 session_id가 있지만,
지금 예제에서는 실제로 short-term memory를 아직 붙이지 않았습니다.

이걸 일부러 남겨둔 이유가 있어요.
사내 도우미를 API 서버로 감싸는 첫 단계에서는
먼저 요청/응답 계약을 안정적으로 잡는 게 중요하거든요.

그리고 바로 다음 단계로는 자연스럽게 이게 옵니다.

• session_id 기준 thread 분리
• short-term memory 붙이기
• checkpointer 붙이기

LangChain short-term memory 문서는 agent의 short-term memory가 agent state의 일부이고, thread별로 분리되며 checkpointer로 persistence할 수 있다고 설명합니다. (LangChain Docs)

즉 지금은 API 뼈대를 먼저 만들고,
그 다음에 대화 상태를 붙이기 좋은 형태로 열어둔 겁니다.

---

7. 이 구조가 실무에서 왜 괜찮은 출발점인가

이건 꽤 중요합니다.

처음부터 너무 많은 걸 넣으면
오히려 어디가 문제인지 안 보여요.

지금 구조는 딱 적당합니다.

장점 1. 역할이 나뉘어 있다

• FastAPI는 HTTP 처리
• agent는 판단과 응답 생성
• tools는 외부 기능 흉내

장점 2. 나중에 교체가 쉽다

• mock tool → 실제 API
• simple search → RAG retriever
• no memory → short-term memory
• sync endpoint → streaming endpoint

장점 3. 프론트 붙이기 쉽다

웹이든 앱이든 /assistant/ask만 호출하면 되니까요.

LangChain과 FastAPI 문서를 같이 보면, 하나는 agent/tool abstraction을 제공하고 다른 하나는 typed API surface를 제공하는 역할 분담이 꽤 잘 맞습니다. (LangChain Docs)

---

8. 실제 운영으로 가면 바로 추가될 것들

지금 코드는 학습용 출발점입니다.
실무에 가면 거의 바로 아래가 붙습니다.

인증

사내 도우미라면 사용자 인증 없이 열어두면 안 됩니다.

권한

모든 사용자가 모든 요청 상태를 보면 안 되겠죠.

로깅

누가 어떤 질문을 했고, 어떤 tool을 불렀는지 남겨야 합니다.

timeout / fallback

내부 API가 느리거나 실패할 수 있습니다.

memory

같은 세션의 이전 대화를 이어가야 할 수 있습니다.

streaming

답변이 길면 점진적으로 보여주는 게 UX가 더 좋습니다.

LangGraph streaming 문서는 실시간 업데이트를 surface하는 streaming system이 있고, latency가 있는 LLM 앱 UX 개선에 중요하다고 설명합니다. LangChain middleware 문서는 summarization, model call limit, tool call limit, fallback 같은 production-oriented 장치도 제공합니다. (LangChain Docs)

그러니까 지금 구조는 완성본이 아니라
확장 가능한 첫 백엔드 골격이라고 보면 맞습니다.

---

9. 왜 FastAPI에서 response model을 꼭 쓰는 게 좋을까

이건 생각보다 별거 아닌 것 같지만 꽤 중요합니다.

그냥 dict 반환해도 됩니다.
근데 response model을 쓰면 좋은 점이 많아요.

• 응답 구조가 문서화됨
• 타입이 명확해짐
• 프론트와 약속이 선명해짐
• 직렬화가 안정적임

FastAPI는 response model을 선언하면 Pydantic으로 JSON 직렬화하고, 이 방식이 보통 직접 JSONResponse를 만지는 것보다 일관되고 성능 면에서도 낫다고 설명합니다. (FastAPI)

사내 도우미처럼 기능이 점점 커질 서비스일수록
처음부터 응답 구조를 명확히 해두는 게 정말 도움이 됩니다.

---

10. 나중에 streaming으로 가려면 어떻게 보이면 좋을까

지금은 일반 JSON 응답만 반환했죠.
근데 대화형 도우미는 스트리밍이 꽤 유용합니다.

특히:

• 답변이 길 때
• tool 호출이 여러 번 있을 때
• 사용자에게 “지금 뭔가 하고 있다”는 감각을 주고 싶을 때

LangGraph streaming 문서는 streaming이 real-time updates를 드러내는 핵심 UX 기능이라고 설명합니다. 즉 사내 도우미가 조금만 커져도 나중엔 /assistant/stream 같은 엔드포인트가 생길 가능성이 큽니다. (LangChain Docs)

다만 처음부터 스트리밍까지 넣으면 구조 이해가 흐려질 수 있어서,
이번 글에서는 일부러 JSON API 한 개로 시작한 겁니다.

---

11. 여기서 바로 해볼 만한 개선

이건 다음 단계로 가기 전, 직접 해보면 좋은 것들입니다.

첫째, search_policy_documents를 진짜 retriever 기반 RAG tool로 바꿔보기.
둘째, get_request_status를 실제 internal API 호출로 바꿔보기.
셋째, AskResponse에 used_tools: list[str] 같은 필드를 추가해보기.
넷째, session_id 기준으로 대화 기록 저장을 붙여보기.

이런 개선은 LangChain의 short-term memory, retriever, structured output, middleware와 자연스럽게 이어집니다. (LangChain Docs)

---

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

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

FastAPI로 LangChain 백엔드를 감싼다는 건, AI 예제를 다른 시스템이 호출할 수 있는 서비스 계약으로 바꾸는 일이다.

이 감각이 생기면
이제부터 코드를 보는 시선이 조금 달라집니다.

“이 예제가 돌아간다”가 아니라
“이 기능을 API로 노출하면 어디서 어떻게 붙일 수 있을까”
이렇게 보이기 시작하거든요.

---

마무리

저는 이 구간이 꽤 좋습니다.

왜냐면 여기서부터는 AI 기능 자체보다
그걸 시스템 안에 어떻게 넣을지를 고민하게 되거든요.

LangChain만 보면 종종 기능 데모처럼 느껴질 수 있어요.
FastAPI만 보면 그냥 평범한 CRUD 서버처럼 보일 수도 있고요.

근데 둘을 붙이는 순간
갑자기 되게 현실적인 그림이 나옵니다.

• 프론트가 붙을 수 있고
• 사내 시스템이 호출할 수 있고
• 인증과 권한을 붙일 수 있고
• 운영과 로깅을 생각하게 되고
• “서비스”라는 단어가 조금 더 실감나기 시작합니다

저는 그 지점이 좋더라고요.
화려하진 않아도, 진짜 만들고 있다는 느낌이 들거든요.

---

다음 글 예고

다음 글에서는
LangChain 백엔드에 대화 상태 붙이기 — session_id, short-term memory, thread 분리를 FastAPI에서 어떻게 다루나
로 이어가겠습니다.

이제부터는 단순한 1회성 질문응답이 아니라,
같은 사용자의 여러 턴 대화를 어떻게 서버에서 이어갈지 본격적으로 다루게 될 거예요.

출처

• LangChain Overview: LangChain은 agent와 LLM 애플리케이션을 위한 higher-level framework이며, agents는 LangGraph 위에서 동작한다고 설명합니다. (LangChain Docs)
• LangChain v1 / create_agent: create_agent는 LangChain 1.0의 standard way to build agents라고 설명합니다. (LangChain Docs)
• LangChain Short-term memory: agent의 short-term memory는 state 일부이며 thread별로 분리되고 checkpointer로 이어갈 수 있다고 설명합니다. (LangChain Docs)
• LangChain RAG 문서: retrieval and generation 단계가 분리되고, 런타임에 관련 데이터를 검색해 모델에 전달하는 흐름을 설명합니다. (LangChain Docs)
• FastAPI 공식 소개: FastAPI는 standard Python type hints 기반의 modern, high-performance API framework라고 설명합니다. (FastAPI)
• FastAPI Request Body: 요청 본문은 Pydantic 모델로 선언하는 방식을 기본으로 설명합니다. (FastAPI)
• FastAPI Response / Response Model: response model을 선언하면 Pydantic 기반 JSON 직렬화와 성능상 이점이 있다고 설명합니다. (FastAPI)
• FastAPI Dependencies: dependency injection system을 제공해 컴포넌트 주입이 쉽다고 설명합니다. (FastAPI)
• FastAPI Tutorial: step-by-step 구조로 기능을 점진적으로 확장하는 공식 튜토리얼을 제공합니다. (FastAPI)
• LangGraph Streaming: 실시간 업데이트를 surface하는 streaming system이 UX 향상에 중요하다고 설명합니다. (LangChain Docs)

LangChain, FastAPI, create_agent, AI Agent, LangChain Backend, Python API, 생성형AI, 백엔드개발, 사내도우미, 주니어개발자