[TIL]FastAPI 파일 브릿지와 LangGraph 기초

1. 학습 주제

FastAPI와 Spring을 HTTP로 연결해 파일 업로드를 처리하는 구조와, LangGraph를 사용해 AI 작업 흐름을 Graph 형태로 구성하는 방법을 학습했다.

핵심 내용은 다음과 같다.

• FastAPI 파일 업로드 API 구현
• Spring에서 MultipartFile을 받아 FastAPI로 전달
• RestClient를 사용한 서버 간 HTTP 통신
• LangChain Chain과 LangGraph Graph 차이
• LangGraph의 State, Node, Edge
• Reducer를 사용한 State 병합
• 조건 분기 라우팅
• RAG 흐름을 Graph로 구성
• 질문 의도에 따라 RAG 또는 일반 답변으로 분기

---

2. FastAPI 파일 업로드 API

FastAPI에서는 파일과 일반 값을 함께 받기 위해 multipart/form-data 형식을 사용한다.

@app.post("/api/v1/files", response_model=FileInfo, status_code=201)
async def upload_file(
    file: Annotated[UploadFile, File()],
    description: Annotated[str, Form(min_length=1, max_length=100)]
):

여기서 역할은 다음과 같다.

코드	의미
UploadFile	업로드된 파일 객체
File()	multipart 요청의 파일 part
Form()	multipart 요청의 일반 입력값
description	파일 설명

즉, 하나의 요청 안에 파일과 설명을 함께 담아 보낼 수 있다.

---

3. 파일 검증 흐름

파일 업로드에서는 아무 파일이나 저장하면 안 된다.
그래서 다음 조건을 검증했다.

• 허용된 Content-Type인지 확인
• 확장자와 Content-Type이 일치하는지 확인
• 파일 크기가 1MB 이하인지 확인
• 원본 파일명에서 경로 정보 제거
• 저장 파일명은 UUID로 생성

ALLOWED_TYPES = {
    "text/plain": ".txt",
    "application/pdf": ".pdf"
}

content = await file.read(MAX_FILE_SIZE + 1)

if len(content) > MAX_FILE_SIZE:
    raise HTTPException(status_code=413, detail="파일은 1MB 이하여야 합니다.")

파일명은 사용자가 보낸 이름을 그대로 쓰지 않고 UUID를 사용했다.

file_id = str(uuid4())
stored_name = f"{file_id}{suffix}"

이렇게 하면 파일명 충돌을 줄이고, 사용자가 보낸 파일명에 의존하지 않는 저장 구조를 만들 수 있다.

---

4. 파일 Metadata 저장

파일 자체는 .txt 또는 .pdf로 저장하고, 파일 정보는 별도의 .json 파일로 저장했다.

metadata_path(file_id).write_text(
    json.dumps(info.model_dump(), ensure_ascii=False, indent=2),
    encoding="utf-8"
)

저장되는 정보는 다음과 같다.

필드	의미
id	파일 고유 ID
original_name	사용자가 업로드한 원본 파일명
content_type	파일 MIME 타입
size	파일 크기
description	파일 설명
stored_name	서버에 저장된 파일명

파일과 metadata를 분리하면 나중에 목록 조회, 다운로드, 삭제 기능을 만들 때 관리하기 쉽다.

---

5. Spring과 FastAPI 파일 브릿지

Spring 서버는 클라이언트로부터 파일을 받고, 그 파일을 다시 FastAPI 서버로 전달한다.

흐름은 다음과 같다.

브라우저
-> Spring Controller
-> Spring RestClient
-> FastAPI 파일 업로드 API
-> FastAPI 저장
-> Spring 응답 반환

중요한 점은 Spring이 Python 함수를 직접 실행하는 것이 아니라, FastAPI 서버에 HTTP 요청을 보낸다는 것이다.

Spring Controller에서는 파일을 다음처럼 받는다.

@PostMapping(consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public ResponseEntity<?> upload(
        @RequestPart("file") MultipartFile file,
        @RequestParam("description") String description
)

그리고 FastApiFileClient에서 FastAPI로 다시 multipart 요청을 만든다.

MultiValueMap<String, Object> body = new LinkedMultiValueMap<>();
body.add("file", new HttpEntity<>(resource, fileHeaders));
body.add("description", description);

---

6. RestClient와 Multipart 전달

Spring에서 외부 API를 호출할 때 RestClient를 사용했다.

return restClient.post()
        .uri("/api/v1/files")
        .contentType(MediaType.MULTIPART_FORM_DATA)
        .body(body)
        .retrieve()
        .body(FileInfo.class);

MultipartFile은 그대로 외부 API에 넣을 수 없기 때문에 ByteArrayResource로 감싸서 전달했다.

ByteArrayResource resource = new ByteArrayResource(file.getBytes()) {
    @Override
    public String getFilename() {
        return file.getOriginalFilename();
    }
};

여기서 getFilename()을 오버라이드하는 이유는 FastAPI 쪽에서 원본 파일명을 인식할 수 있게 하기 위해서다.

---

7. 외부 API 예외 처리

FastAPI에서 파일 형식이 잘못되었거나 서버 연결이 실패할 수 있다.
Spring에서는 이 경우를 나누어 처리했다.

catch(RestClientResponseException e) {
    return ResponseEntity.status(e.getStatusCode())
            .body(Map.of(
                    "code", "FASTAPI_REJECTED",
                    "message", e.getResponseBodyAsString()
            ));
}

catch(IOException | RestClientException e) {
    return ResponseEntity.status(HttpStatus.BAD_GATEWAY)
            .body(Map.of(
                    "code", "UPLOAD_FAILED",
                    "message", "파일 전달에 실패했습니다."
            ));
}

정리하면 다음과 같다.

상황	응답
FastAPI가 파일을 거절	FastAPI의 상태 코드 전달
파일 읽기 실패 또는 통신 실패	502 Bad Gateway
FastAPI 서버 연결 불가	503 Service Unavailable

외부 서버와 연동할 때는 성공 흐름뿐 아니라 실패 흐름도 반드시 고려해야 한다.

---

8. LangChain Chain과 LangGraph Graph 차이

LangChain Chain은 작업 순서가 고정된 파이프라인에 적합하다.

retrieve -> generate

예를 들어 항상 검색 후 답변 생성만 한다면 Chain으로 충분하다.

하지만 질문에 따라 실행 경로가 달라지거나, 중간 상태를 여러 노드가 공유하거나, 분기와 반복이 필요하면 LangGraph가 더 적합하다.

START
-> classify
-> 조건에 따라 budget / destination / general
-> END

정리하면 다음과 같다.

구분	적합한 상황
Chain	순서가 고정된 단순 흐름
Graph	분기, 반복, 조건 처리, 상태 공유가 필요한 흐름

---

9. LangGraph의 State, Node, Edge

LangGraph는 작업 흐름을 State, Node, Edge로 구성한다.

개념	의미
State	여러 Node가 공유하는 데이터
Node	실제 작업을 수행하는 함수
Edge	Node 사이의 실행 순서
START	그래프 시작 지점
END	그래프 종료 지점

예시는 다음과 같다.

class TripState(TypedDict):
    place: str
    travelers: int
    ticket_price: int
    total_price: int
    message: str
    path: list[str]

Node는 State 전체를 반환하지 않아도 된다.
자신이 변경할 값만 dict로 반환하면 LangGraph가 기존 State에 합쳐준다.

def calculate_total(state: TripState) -> dict:
    total = state["travelers"] * state["ticket_price"]
    return {"total_price": total}

---

10. Reducer

기본적으로 같은 State 필드를 여러 번 업데이트하면 새 값으로 덮어써질 수 있다.
하지만 리스트처럼 계속 누적해야 하는 값은 덮어쓰기보다 합치기가 필요하다.

이때 사용하는 것이 Reducer이다.

class ResearchState(TypedDict):
    topic: str
    notes: Annotated[list[str], operator.add]

operator.add를 사용하면 여러 Node가 반환한 notes가 덮어써지지 않고 이어 붙는다.

destination note
+ budget note
+ meal note
+ transport note
= 최종 notes

Reducer는 여러 Node가 같은 State 필드를 업데이트할 때 “어떻게 합칠지” 정하는 규칙이다.

---

11. Conditional Routing

모든 질문에 같은 작업을 실행할 필요는 없다.
예산 질문이면 예산 노드로, 장소 질문이면 장소 노드로, 일반 질문이면 일반 노드로 보내는 것이 더 효율적이다.

먼저 질문을 분류하는 Router Node를 만든다.

def classify(state: RouteState) -> dict:
    if "입장료" in state["question"]:
        route = "budget"
    elif "장소" in state["question"]:
        route = "destination"
    else:
        route = "general"
    return {"route": route}

그리고 add_conditional_edges로 다음 노드를 선택한다.

builder.add_conditional_edges("classify", choose_route)

흐름은 다음과 같다.

START
-> classify
-> budget 또는 destination 또는 general
-> END

조건 분기를 사용하면 질문 의도에 맞는 작업만 실행할 수 있다.

---

12. RAG Graph

RAG도 LangGraph로 구성할 수 있다.

START
-> retrieve
-> generate
-> END

각 Node의 역할은 다음과 같다.

Node	역할
retrieve	VectorStore에서 질문과 관련 있는 문서 검색
generate	검색된 context를 근거로 LLM 답변 생성

def retrieve(state: RagState) -> dict:
    documents = store.similarity_search(state["question"], k=2)
    return {
        "context": [document.page_content for document in documents],
        "path": ["retrieve"]
    }

def generate(state: RagState) -> dict:
    prompt = (
        "아래 문서만 근거로 답하세요.\n"
        f"[문서]\n{context}\n[질문]\n{state['question']}"
    )
    answer = model.invoke(prompt).content
    return {"answer": answer}

이 구조의 장점은 RAG가 틀렸을 때 어느 단계가 문제인지 확인하기 쉽다는 점이다.

검색된 context가 이상함 -> Retrieval 문제
context는 맞는데 답변이 이상함 -> Generation 문제

---

13. Question Router

질문에 따라 RAG를 사용할지 일반 답변을 사용할지 나누는 구조도 학습했다.

START
-> classify
-> RAG 질문이면 retrieve -> rag_answer
-> 일반 질문이면 general
-> END

예를 들어 다음 질문은 RAG가 필요하다.

아쿠아리움 입장료는 얼마야?
무료 산책 장소는 어디야?

이 질문들은 수업용 문서에 있는 정보를 근거로 답해야 하므로 retrieve를 거친다.

반면 다음 질문은 일반 답변으로 처리할 수 있다.

여행 가방을 가볍게 싸는 방법은?
아이 여행 준비물은 무엇이야?

이런 질문은 특정 문서 검색이 꼭 필요하지 않으므로 general 노드로 보낸다.

이렇게 하면 불필요한 Embedding 검색 비용을 줄이고, 질문 성격에 맞는 처리 흐름을 만들 수 있다.

---

14. 핵심 정리

1. FastAPI는 UploadFile, File, Form으로 파일과 일반 값을 함께 받을 수 있다.
2. 파일 업로드 요청은 multipart/form-data 형식을 사용한다.
3. 파일 저장 전 Content-Type, 확장자, 크기 검증이 필요하다.
4. 저장 파일명은 UUID를 사용하면 충돌을 줄일 수 있다.
5. Spring은 RestClient로 FastAPI API를 호출할 수 있다.
6. Spring이 Python 함수를 직접 실행하는 것이 아니라 HTTP 요청으로 FastAPI와 통신한다.
7. MultipartFile을 외부 API로 전달할 때는 ByteArrayResource와 MultiValueMap을 사용할 수 있다.
8. 외부 API 연동에서는 RestClientResponseException, RestClientException 등 실패 상황을 나누어 처리해야 한다.
9. LangChain Chain은 고정된 순서의 작업에 적합하다.
10. LangGraph는 분기, 반복, 상태 공유가 필요한 작업 흐름에 적합하다.
11. State는 그래프 전체에서 공유되는 데이터이다.
12. Node는 State를 읽고 일부 값을 업데이트하는 작업 단위이다.
13. Edge는 Node 사이의 실행 순서를 정의한다.
14. Reducer는 같은 State 필드를 여러 Node가 업데이트할 때 병합 방식을 정한다.
15. Conditional Routing은 질문 의도에 따라 다음 실행 노드를 선택하는 방식이다.
16. RAG Graph는 검색 단계와 답변 생성 단계를 분리해 디버깅하기 쉽다.
17. Question Router를 사용하면 RAG가 필요한 질문과 일반 질문을 나누어 처리할 수 있다.