LangChain + FastAPI로 API 서버 만들기

📌 3줄 요약
· LangChain 체인을 노트북 안에만 두지 않고, FastAPI로 감싸면 누구나 호출할 수 있는 API 서버가 됩니다.
· 핵심은 요청(request) → 체인 실행 → 응답(response)의 흐름이며, FastAPI의 엔드포인트 함수 안에서 체인을 invoke하는 것이 전부입니다.
· 입력·출력 형식은 Pydantic 모델로 정의하고, 긴 답변은 스트리밍으로 흘려보내면 실제 서비스에 가깝게 다듬을 수 있습니다.
🌐 HTTP 요청 → 🚪 FastAPI 엔드포인트 → 🔗 체인 실행(invoke) → ✅ JSON 응답

지금까지는 LangChain 체인을 주로 노트북이나 터미널 안에서 직접 실행해 봤습니다. 그런데 만든 체인을 웹·앱·다른 서버에서 가져다 쓰려면, 결국 "밖에서 호출할 수 있는 창구"가 필요합니다. 그 창구가 바로 API 서버이고, 파이썬에서 이를 가장 가볍게 만들 수 있는 도구가 FastAPI입니다. 오늘은 완성한 체인을 FastAPI로 감싸 호출 가능한 API로 바꾸는 과정을 차근차근 따라가 보겠습니다.

🧩 왜 체인을 API로 감싸야 할까

잘 만든 체인이라도, 코드 안에서만 돌아간다면 그 코드를 실행할 수 있는 사람만 쓸 수 있습니다. 프런트엔드 개발자나 모바일 앱, 또는 다른 백엔드 서비스가 이 체인을 쓰고 싶어도 직접 파이썬을 돌릴 수는 없는 경우가 대부분입니다.

그래서 체인을 HTTP로 호출할 수 있는 형태로 열어 두는 것이 중요합니다. 한번 API로 감싸 두면, 어떤 언어·어떤 환경에서도 주소로 요청만 보내면 체인의 결과를 받아 갈 수 있습니다. 즉 API 서버는 체인과 바깥 세상을 이어 주는 다리인 셈입니다.

체인을 API로 감싸는 일은 "혼자 쓰던 기능"을 "누구나 호출할 수 있는 서비스"로 바꾸는 작업입니다.

🚪 FastAPI와 체인이 만나는 지점

FastAPI와 LangChain의 결합은 생각보다 단순합니다. FastAPI가 맡는 일과 체인이 맡는 일을 카드로 나눠 보면 역할이 또렷해집니다.

🚪 FastAPI
외부 요청을 받고, 입력을 검증하고, 체인의 결과를 JSON으로 돌려주는 창구 역할을 합니다.
🔗 LangChain 체인
전달받은 입력으로 실제 작업(LLM 호출 등)을 수행해 답을 만들어 냅니다.
📨 엔드포인트 함수
둘을 잇는 접점으로, 요청에서 꺼낸 값을 체인에 넘기고 결과를 받아 응답으로 포장합니다.

정리하면, FastAPI가 요청을 받아 엔드포인트 함수로 넘기고, 그 함수 안에서 체인을 실행해 답을 얻은 뒤, 다시 FastAPI가 응답으로 포장해 돌려줍니다. 결국 우리가 새로 쓸 코드는 "엔드포인트 함수 안에서 체인을 invoke하는 몇 줄"이 중심이 됩니다.

📋 입력·출력 형식 정의하기

API를 열 때 가장 먼저 정해야 할 것은 "어떤 형태의 데이터가 오가는가"입니다. FastAPI는 Pydantic 모델로 이 형식을 선언하도록 되어 있습니다. 입력 모델을 정의해 두면, 형식에 맞지 않는 요청은 FastAPI가 알아서 걸러 주고, 잘못된 부분을 친절한 오류 메시지로 알려 줍니다.

예를 들어 질문을 받아 답을 돌려주는 API라면, 입력에는 question이라는 문자열 필드가, 출력에는 answer라는 문자열 필드가 있으면 충분합니다. 이렇게 형식을 미리 선언해 두면 코드가 명확해지고, 자동 생성되는 API 문서에도 그대로 반영됩니다.

⚙️ 코드로 살펴보기

이제 실제로 체인을 FastAPI로 감싸 보겠습니다. 먼저 간단한 체인과 입력·출력 모델을 준비합니다. 체인은 앞서 다뤄 온 "프롬프트 | 모델 | 파서" 형태 그대로입니다.

from fastapi import FastAPI
from pydantic import BaseModel
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 1) 체인 준비
prompt = ChatPromptTemplate.from_template("질문에 한국어로 답해라.\n질문: {question}")
chain = prompt | ChatOpenAI(temperature=0) | StrOutputParser()

# 2) 입력·출력 형식 정의
class AskRequest(BaseModel):
    question: str

class AskResponse(BaseModel):
    answer: str

다음으로 FastAPI 앱을 만들고, /ask라는 엔드포인트를 추가합니다. 핵심은 함수 안에서 요청으로 받은 값을 체인에 넘기고, 그 결과를 응답 모델에 담아 돌려주는 부분입니다.

app = FastAPI()

@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest):
    # 요청에서 꺼낸 question을 체인에 넘긴다
    result = await chain.ainvoke({"question": req.question})
    return AskResponse(answer=result)

이 파일을 main.py로 저장했다면, 아래 명령으로 서버를 띄울 수 있습니다. 그러면 /ask 주소로 질문을 POST하는 순간 체인이 실행되고, 답이 JSON으로 돌아옵니다.

# 터미널에서 서버 실행
uvicorn main:app --reload

# 브라우저에서 http://127.0.0.1:8000/docs 를 열면
# 자동 생성된 API 문서로 바로 테스트할 수 있다

여기서 눈여겨볼 점은, FastAPI의 엔드포인트를 async 함수로 두고 체인도 ainvoke(비동기 실행)로 호출했다는 것입니다. LLM 호출처럼 기다림이 많은 작업은 비동기로 처리해야, 한 요청이 끝나길 기다리는 동안 다른 요청도 함께 받을 수 있습니다.

📡 스트리밍 응답 다루기

LLM의 답은 길어질 수 있어서, 완성될 때까지 통째로 기다리면 사용자가 한참 빈 화면을 보게 됩니다. 이럴 때는 답을 조각조각 흘려보내는 스트리밍이 유용합니다. FastAPI에서는 StreamingResponse를 쓰고, 체인 쪽에서는 astream으로 토큰을 하나씩 받아 그대로 내보내면 됩니다.

from fastapi.responses import StreamingResponse

@app.post("/ask-stream")
async def ask_stream(req: AskRequest):
    async def gen():
        # 체인이 만드는 토큰을 받는 대로 흘려보낸다
        async for chunk in chain.astream({"question": req.question}):
            yield chunk
    return StreamingResponse(gen(), media_type="text/plain")

이렇게 하면 답이 생성되는 즉시 조금씩 전달되어, 사용자는 글자가 타이핑되듯 차오르는 화면을 보게 됩니다. 체감 속도가 크게 달라지는 부분이라, 대화형 서비스라면 거의 기본으로 챙기는 방식입니다.

🔧 실무 팁

  • 비동기 호출을 기본으로 — LLM 요청은 대기가 길므로 엔드포인트를 async + ainvoke로 두어야, 동시에 들어오는 여러 요청을 효율적으로 받을 수 있습니다.
  • API 키는 코드 밖에 — OpenAI 키 같은 비밀값은 소스에 직접 쓰지 말고 환경 변수로 두어, 코드가 공유돼도 키가 새지 않게 합니다.
  • 자동 문서를 활용 — FastAPI는 /docs에 대화형 문서를 자동으로 만들어 줍니다. 입력·출력 모델을 잘 정의할수록 이 문서가 더 정확해집니다.
  • 오류 처리도 미리 — 체인 실행이 실패할 수 있으니, 예외를 받아 적절한 상태 코드와 메시지로 돌려주면 호출하는 쪽이 원인을 파악하기 쉽습니다.

❓ 자주 묻는 질문 (FAQ)

Q. 꼭 FastAPI여야 하나요? Flask로는 안 되나요?
A. Flask로도 가능합니다. 다만 FastAPI는 비동기 처리와 Pydantic 기반 입력 검증, 자동 문서 생성을 기본으로 지원해, LLM처럼 대기가 긴 작업을 API로 여는 데 특히 잘 맞습니다.

Q. LangServe라는 것도 들었는데, 이것과 뭐가 다른가요?
A. LangServe는 체인을 더 적은 코드로 API에 붙이도록 도와주는 도구입니다. 내부적으로는 역시 FastAPI 위에서 동작하므로, 오늘처럼 직접 엔드포인트를 짜는 방식을 이해해 두면 LangServe의 동작도 훨씬 수월하게 받아들일 수 있습니다.

Q. 요청마다 체인을 새로 만들어야 하나요?
A. 아닙니다. 체인은 서버가 뜰 때 한 번만 만들어 두고, 각 요청에서는 그 체인을 invoke만 하면 됩니다. 매 요청마다 다시 만들면 불필요한 비용과 지연이 생깁니다.

마무리

정리하면, LangChain 체인을 FastAPI로 감싸는 일은 요청 → 체인 실행 → 응답이라는 간단한 흐름을 코드로 옮기는 작업이었습니다. 핵심은 엔드포인트 함수 안에서 체인을 invoke하는 것이었고, 입력·출력은 Pydantic 모델로 명확히 선언해 FastAPI가 검증과 문서화를 대신 맡도록 했습니다. 또한 LLM 호출은 대기가 길기 때문에 비동기 처리를 기본으로 두고, 긴 답변은 스트리밍으로 흘려보내 체감 속도를 끌어올릴 수 있다는 점도 살펴봤습니다. 이렇게 한 번 API로 열어 두면, 혼자 쓰던 체인이 누구나 호출할 수 있는 서비스로 거듭납니다. 다음 편에서는 이렇게 만든 API에 인증과 사용량 관리를 더해 실제 운영에 가깝게 다듬는 이야기로 이어가 보겠습니다.

댓글

댓글 쓰기

이 블로그의 인기 게시물

LCEL — LangChain 표현식으로 체인 깔끔하게 잇기

📌 3줄 요약 · LCEL(LangChain Expression Language) 은 여러 단계를 파이프 기호(|) 로 이어 하나의 체인으로 만드는 방식이에요. · 프롬프트 → 모델 → 출력 파서처럼 흐름이 그대로 코드에 보여서 읽기 쉽고 수정도 간편합니다. · 스트리밍·배치·비동기 같은 기능을 따로 구현하지 않아도 체인 전체에서 바로 쓸 수 있습니다. 📖 목차 LCEL이 왜 필요할까 파이프(|)로 잇는다는 발상 체인의 기본 3단 구성 코드로 보는 LCEL LCEL이 공짜로 주는 기능들 실무 팁 자주 묻는 질문(FAQ) 📝 프롬프트 → 🤖 모델 → 🔧 출력 파서 → ✅ 결과 LangChain으로 뭔가를 만들다 보면, 프롬프트를 만들고 모델에 넣고 그 결과를 다시 다듬는 여러 단계 가 반복됩니다. 예전에는 이걸 함수로 하나씩 호출하며 이어 붙였는데, 코드가 길어질수록 흐름을 알아보기 어려웠어요. 이 불편을 깔끔하게 정리한 것이 바로 LCEL(LangChain Expression Language) 입니다. 오늘은 LCEL이 무엇이고 왜 편한지 차근차근 살펴보겠습니다. 🤔 LCEL이 왜 필요할까 LLM 앱은 대부분 여러 단계가 순서대로 이어지는 구조 입니다. 사용자 질문을 받아 프롬프트 틀에 채우고, 그걸 모델에 보내고, 모델이 돌려준 답을 원하는 형태로 정리하는 식이죠. 각 단계를 따로 함수로 부르면 이렇게 됩니다. 단계가 두세 개일 땐 괜찮지만, 중간에 검색이나 조건 분기가 끼면 어디서 무엇이 넘어가는지 한눈에 파악하기 어려워집니다. 게다가 스트리밍이나 병렬 처리를 넣으려면 단계마다 코드를 또 손봐야 하죠. LCEL의 목표는 간단합니다. "단계를 이어 붙이는 일"을 눈에 보이게, 그리고 한 번만 만들자는 것입니다. 🔗 파이프(|)로 잇는다는 발상 LCEL의 핵심은 파이프 기호 | 하나입니다. 리눅스 터미널에서 명령어1 | 명령어2 로 앞 결과를 뒤로 ...
자세한 내용 보기

Streaming — 답변을 실시간으로 흘려보내기

📌 3줄 요약 · 스트리밍(Streaming) 은 모델의 답변을 다 만들어진 뒤 한꺼번에 주는 대신, 생성되는 대로 조금씩 흘려보내는 방식이에요. · LangChain에서는 invoke 대신 stream 을 쓰면, 답변이 토큰 단위로 순서대로 도착합니다(타이핑 효과). · 체감 대기 시간이 크게 줄어 챗봇처럼 즉각 반응하는 UX 를 만들 때 특히 유용합니다. 📖 목차 스트리밍이 왜 필요할까 한 번에 vs 조금씩, 무엇이 다른가 토큰 단위로 흘려보낸다는 것 코드로 보는 stream stream vs invoke vs batch 실무 팁 자주 묻는 질문(FAQ) 🙋 질문 → 🤖 모델 생성 → 💧 토큰이 하나씩 → 🖥️ 화면에 실시간 챗봇에 질문을 던졌을 때, 답이 한 글자씩 타이핑되듯 나타나는 화면을 본 적 있을 거예요. 이건 답변이 다 완성될 때까지 기다렸다가 한꺼번에 보여 주는 게 아니라, 모델이 글자를 만들어 내는 즉시 화면으로 흘려보내기 때문에 가능한 일입니다. 이 방식을 스트리밍(Streaming) 이라고 부릅니다. 오늘은 LangChain에서 스트리밍이 무엇이고 왜 유용한지, 그리고 어떻게 쓰는지 차근차근 살펴보겠습니다. 🤔 스트리밍이 왜 필요할까 LLM은 답변을 앞에서부터 순서대로 한 조각씩 생성합니다. 긴 답변일수록 전체가 완성되기까지 시간이 걸리는데, 완성될 때까지 화면에 아무것도 안 보이면 사용자는 "멈춘 건가?" 하고 답답해집니다. 스트리밍은 이 기다리는 시간을 체감상 크게 줄여 줍니다. 첫 글자가 나오기 시작하는 순간부터 화면에 하나씩 채워지니, 실제 전체 생성 시간이 같아도 훨씬 빠르게 느껴지죠. 특히 대화형 챗봇처럼 반응 속도가 중요한 서비스 에서는 거의 필수적인 요소입니다. 스트리밍의 핵심은 "답을 다 만들 때까지 기다리지 말고, 만들어지는 대로 바로 보여 주자" 는 것입니다. ⚖️ 한 번에 vs 조금씩...
자세한 내용 보기

Callback와 로깅 — LangChain 내부 들여다보기

📌 3줄 요약 · 콜백(Callback) 은 LangChain이 실행되는 중간중간 특정 시점마다 내가 지정한 함수를 자동으로 불러 주는 장치예요. · 이 콜백을 이용하면 체인이 언제 시작하고, 어떤 프롬프트가 들어가고, 어떤 답이 나왔는지 를 실시간으로 들여다보고 로깅 할 수 있습니다. · 덕분에 "왜 이런 답이 나왔지?"를 추적하는 디버깅과 모니터링 이 훨씬 쉬워집니다. 📖 목차 콜백이 왜 필요할까 콜백은 언제 불릴까 — 이벤트라는 개념 로깅과 콜백은 어떻게 연결되나 코드로 보는 콜백 핸들러 대표적인 콜백 이벤트 한눈에 실무 팁 자주 묻는 질문(FAQ) 🚀 체인 시작 → 🔔 이벤트 발생 → 🪝 콜백 호출 → 📝 기록·로깅 체인을 만들어 실행하면, 겉으로는 질문을 넣으면 답이 나오는 단순한 상자처럼 보입니다. 하지만 그 상자 안에서는 프롬프트가 조립되고, 모델이 호출되고, 출력이 정리되는 여러 단계가 순서대로 벌어지고 있어요. 이 내부 과정을 밖에서 들여다볼 수 있게 해 주는 장치가 바로 콜백(Callback) 입니다. 오늘은 콜백이 무엇이고 왜 유용한지, 그리고 이를 활용한 로깅 이 어떻게 이뤄지는지 차근차근 살펴보겠습니다. 🤔 콜백이 왜 필요할까 체인이 예상과 다른 답을 내놓았을 때, 우리가 볼 수 있는 건 보통 최종 결과 하나뿐 입니다. 그런데 문제의 원인은 대개 중간에 있어요. 실제로 어떤 프롬프트가 모델에 들어갔는지 , 모델이 어떤 원문을 돌려줬는지 를 모르면 원인을 짚기 어렵습니다. 콜백은 이 보이지 않던 중간 과정을 밖으로 꺼내 주는 역할을 합니다. 체인이 시작될 때, 모델이 호출될 때, 답이 완성될 때처럼 중요한 순간마다 내가 등록해 둔 함수가 자동으로 실행되죠. 그래서 실행 흐름을 기록하고 추적하는 일 이 한결 수월해집니다. 콜백의 핵심은 "실행 중간의 특정 순간마다, 내가 원하는 코드를 끼워 넣을 수 있게 하자" 는 것입...
자세한 내용 보기