출처: Dev.to

pydantic‑ai와 FastAPI로 멀티‑에이전트 프롬프트 엔지니어링 런북을 만든 방법

대부분의 AI 툴링 팀은 결국 같은 벽에 부딪힙니다. 다섯 가지 서로 다른 프롬프트 패턴이 Notion 문서, Slack 스레드, 누군가의 로컬 Python 파일에 흩어져 있고, 출력 형식에 대한 합의가 없기 때문이죠. SWOT 분석 프롬프트는 가끔 마크다운을, 가끔 JSON을 반환하고, 코드 리뷰 프롬프트는 그냥 텍스트만 덤프합니다. 프로덕션에서 뭔가가 깨지면 실제로 실행되고 있던 프롬프트 버전을 찾는 데 40분을 허비하게 됩니다.

이 글에서는 pydantic‑ai, FastAPI, 그리고 구조화된 Pydantic 출력을 활용해 이 문제를 해결하는 아키텍처를 소개합니다. 결과물은 하나의 배포 가능한 서비스로, SWOT 분석, 소셜 포스트 생성, 코드 리뷰, 다중 포맷 요약, 의사결정 프레임워크 등을 모두 처리하고 타입이 지정되고 검증된 응답을 반환하는 프롬프트 엔지니어링 런북입니다.

구체적인 시나리오

다섯 명 이상의 엔지니어가 있는 팀을 예로 들어보겠습니다.

1. 누군가가 Jupyter 노트북에 유용한 SWOT 분석 프롬프트를 작성합니다. 잘 동작합니다.
2. 팀원이 이를 FastAPI 라우트에 복사해 몇 마디만 바꾸고 모델 이름을 하드코딩합니다.
3. 세 달 뒤, 또 다른 사람이 약간 다른 버전을 사용하는 Slack 봇을 만들었습니다.

이제 프로덕션에는 서로 다른 계약이 전혀 없는 세 개의 SWOT 분석기가 존재합니다.

하위 시스템이 깨지는 이유는 한 버전은 strengths를 리스트로 반환하고, 다른 버전은 콤마‑구분 문자열로 반환하기 때문입니다. 코드 리뷰 프롬프트는 원시 텍스트만 반환하므로 프론트엔드에서 정규식으로 파싱해야 합니다. 모델을 업그레이드하면 어느 프롬프트가 조용히 성능 저하를 일으킬지 전혀 알 수 없습니다.

Slack을 진실의 원천으로 삼는 팀은 특히 이 문제에 취약합니다. 컨텍스트는 메모리에서 사라지는 스레드에 살아있고, 결정 사항은 묻혀버리며, 구조화된 인사이트를 추출하려면 수작업을 하거나 아무도 유지보수하지 않는 비공식 스크립트에 의존하게 됩니다. “우리 AI 출력은 이렇게 생겼다”는 단일 장소가 없기 때문에 혼란은 배가됩니다.

해결책은 더 나은 프롬프트가 아니라, 프롬프트와 시스템 사이에 타입이 지정된 계약 레이어를 두는 것입니다.

핵심 아이디어는 간단합니다. 런북에 있는 모든 에이전트는 출력 타입으로 Pydantic 모델을 사용합니다. pydantic‑ai가 LLM 호출 경계에서 그 계약을 강제합니다. FastAPI는 각 에이전트를 타입이 지정된 요청·응답 바디를 가진 엔드포인트로 노출합니다.

왜 pydantic‑ai인가?

• LangChain은 가장 흔히 비교되는 대안입니다. 출력 파서와 구조화된 출력 지원을 제공하지만, 추상화 레이어가 두껍습니다. 파싱 실패를 디버깅하려면 여러 내부 체인 객체를 추적해야 합니다. 팀 전체가 유지보수해야 하는 런북에서는 이런 불투명성이 큰 부담이 됩니다.
• plain requests + instructor도 비슷한 접근법이며 충분히 유효한 선택입니다. 다만 pydantic‑ai는 에이전트 수준의 재시도와 툴 지원을 기본 제공하는데, 이는 컨텍스트 검색이나 다단계 추론을 추가할 때 큰 장점이 됩니다.
• OpenAI 구조화 출력은 동작하지만 공급자에 종속됩니다. pydantic‑ai는 공급자에 구애받지 않으므로 OpenAI에서 Anthropic, 로컬 모델 등으로 바꾸는 것이 설정만 바꾸면 되며 코드 재작성은 필요 없습니다.

신뢰성을 보장하는 핵심 설계 결정

모든 에이전트는 result_type이 문자열이 아니라 Pydantic 모델로 정의됩니다. pydantic‑ai는 출력이 검증에 실패하면 자동으로 LLM 호출을 재시도합니다. 검증 피드백을 프롬프트에 다시 넣어 재시도하는 메커니즘은 순수 프롬프트 엔지니어링만으로는 얻을 수 없는 기능입니다.

FastAPI 레이어는 들어오는 요청에 HTTP‑레벨 검증을, 나가는 응답에 직렬화를 추가합니다. 모든 요청·응답이 타입이 지정되어 있으므로 프론트엔드, Slack 봇, CI 파이프라인 모두 동일한 계약을 사용합니다.

중앙 패턴

런북에 포함되는 모든 로직은 다음과 같은 형태를 따릅니다.

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from fastapi import FastAPI, HTTPException

# 1. 출력 계약 정의
class SWOTAnalysis(BaseModel):
    strengths: list[str] = Field(description="내부 긍정 요인")
    weaknesses: list[str] = Field(description="내부 부정 요인")
    opportunities: list[str] = Field(description="외부 긍정 요인")
    threats: list[str] = Field(description="외부 부정 요인")
    summary: str = Field(description="두 문장으로 요약한 임원용 개요")

# 2. 입력 정의
class SWOTRequest(BaseModel):
    context: str = Field(description="분석할 비즈니스 혹은 제품 컨텍스트")
    focus_area: str | None = Field(default=None, description="선택적 도메인 초점")

# 3. 계약을 강제하는 result_type으로 에이전트 생성
swot_agent = Agent(
    model="openai:gpt-4o",
    result_type=SWOTAnalysis,
    system_prompt=(
        "당신은 전략 분석가입니다. 제공된 컨텍스트를 분석하고 "
        "구조화된 SWOT 분석을 반환하세요. 구체적이고 실행 가능하게 작성하고, "
        "각 리스트는 3~5개의 항목을 포함해야 합니다."
    ),
)

app = FastAPI()

# 4. 타입이 지정된 FastAPI 엔드포인트로 노출
@app.post("/analyse/swot", response_model=SWOTAnalysis)
async def analyse_swot(request: SWOTRequest) -> SWOTAnalysis:
    prompt = request.context
    if request.focus_area:
        prompt = f"Focus area: {request.focus_area}\n\nContext: {request.context}"

    try:
        result = await swot_agent.run(prompt)
        return result.data
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

각 부분이 하는 일과 중요한 이유

• result_type=SWOTAnalysis는 핵심 라인입니다. 이것이 pydantic‑ai에 구조화된 출력 모드를 사용하고, 응답을 Pydantic 스키마에 맞춰 검증하도록 지시합니다. LLM이 잘못된 JSON이나 누락된 필드를 반환하면 자동으로 재시도합니다.
• FastAPI 라우트의 response_model=SWOTAnalysis 덕분에 OpenAPI 문서가 실제 출력 타입을 기반으로 자동 생성됩니다. 프론트엔드 개발자는 프롬프트를 읽지 않아도 어떤 필드가 반환되는지 정확히 알 수 있습니다.
• result.data는 검증된 Pydantic 인스턴스를 바로 반환합니다. 별도의 JSON 파싱이나 .get() 호출이 필요 없습니다.

같은 패턴이 코드 리뷰, 소셜 포스트 생성, 다중 포맷 요약, 의사결정 프레임워크 등 런북에 포함된 모든 에이전트에 반복됩니다. 각각은 서로 다른 Pydantic 모델과 시스템 프롬프트를 갖지만 구조는 동일합니다.

외부 데이터와 연결될 때 진가가 발휘됩니다

대부분의 팀에게 가장 큰 효과를 주는 통합은 Slack입니다. 데이터 흐름은 다음과 같습니다.

Slack 채널/스레드
    → Slack API (conversations.history 또는 웹훅)
    → 런북의 추출 엔드포인트
    → 요약기 또는 SWOT 에이전트
    → 구조화된 출력이 Postgres에 저장되거나 Slack으로 반환

Slack 통합 예시:

from slack_sdk import WebClient

slack_client = WebClient(token=settings.slack_bot_token)

def extract_thread_context(channel_id: str, thread_ts: str) -> str:
    response = slack_client.conversations_replies(
        channel=channel_id,
        ts=thread_ts
    )
    messages = response["messages"]
    return "\n".join(
        f"{msg.get('username', 'user')}: {msg['text']}"
        for msg in messages
    )