Agent AI Skeleton

LangGraph로 만든 에이전트 서비스 골격입니다. 외부 AI 모델 API를 도구로 부르고, FastAPI로 감싸 뒀습니다.

프로젝트를 새로 시작할 때마다 같은 배선을 반복하는 게 지겨워서 정리해 둔 저장소입니다. 도메인 로직은 비워 뒀으니 모델 어댑터와 도구만 갈아끼우면 됩니다.

대신 한 가지는 공들여 만들었습니다. 도구가 실패했을 때 에이전트가 아무 말이나 지어내지 않는 것. 이게 생각보다 자주 터지는 문제입니다.

핵심 흐름

사용자 → FastAPI(/chat · /chat-stream) → AgentFacade → AgentOrchestrator(LangGraph)
            의도분류 → 도구계획 ─┬ 도구실행 ─┬ 답변생성
                                  │           ├ 재시도(전량 실패, 한도 내)
                                  │           └ 근거없음 거부
                                  └ (도구 불필요) 답변생성 / (근거 필수·LLM 장애) 거부

도구 공급: LocalToolProvider(포트 → 외부 모델 API) / McpToolProvider(MCP 서버)

코어(aiagent/)는 DB도, 웹 프레임워크도, 모델 구현도 모릅니다. 필요한 건 전부 진입점(main.py)에서 만들어 넘겨줍니다.

폴더 구조

main.py                         조립 루트 (lifespan에서 모든 의존성 생성·주입)
aiagent/                        프레임워크 독립 코어
├─ config/{schemas,loader}.py   추론 LLM 설정 (LLM_* 환경변수)
├─ ports/inference_model.py     ★ 외부 모델 API 호출 경계(포트)
├─ ports/tool_provider.py       ★ 도구 공급 경계(포트) — 로컬/MCP 공통
├─ tools/tool_registry.py       포트를 도구로 노출 (build_tools · LocalToolProvider)
├─ tools/tool_contract.py       ★ 도구 이름, 의도별 시나리오, 반환 검증 모델
├─ tools/response_validator.py  도구 응답 계약 검증
├─ agents/                      orchestrator · facade · prompt_builder · state
├─ schemas/agent_io.py          계층 간 데이터 계약
└─ core/factory.py              assemble_orchestrator(...)
app/                            FastAPI 어댑터 (transport 계층)
├─ api/routes.py                /health · /chat · /chat-stream(SSE)
├─ core/config.py               서버, 모델 API, 도구 공급원, 이력 한도 설정
├─ core/dependencies.py         DI
├─ core/observability.py        request_id 전파, 처리시간 로깅, 오류 응답
├─ agent/conversation_store.py  대화 이력 보관 경계 (길이 제한, LRU 방출)
├─ adapters/http_inference_client.py  ★ 모델 API 포트의 HTTP 구현
└─ adapters/mcp_tool_provider.py      ★ 도구 포트의 MCP 구현
frontend/index.html             데모 채팅 UI (/ui) — 단건·SSE 스트리밍 호출 예제
docs/adr/                       왜 이렇게 만들었는지에 대한 기록
tests/                          코어 로직 · 근거 가드 · 장애 내성 · API

★ 표시가 붙은 게 실제 프로젝트에서 갈아끼우게 될 파일입니다.

실행

Python 3.11 이상이 필요합니다.

cp .env.example .env                    # 값 채우기

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"                 # MCP 도구도 쓰려면: pip install -e ".[dev,mcp]"

python main.py                          # http://localhost:8000/ui

uv를 쓴다면 uv sync --extra dev(+ --extra mcp)로 대신할 수 있습니다. 다만 uv.lock을 커밋해 두지 않아서, 어느 쪽이든 버전은 설치할 때마다 새로 해석됩니다. 고정이 필요하면 락 파일을 만들어 커밋하세요.

다음 값이 잘못돼 있으면 기동 단계에서 바로 실패합니다. 일부러 그렇게 뒀습니다. 설정이 빠진 채로 뜬 다음 첫 요청에서 죽는 것보다는 나으니까요.

• LLM_API_KEY, MODEL_API_KEY가 비어 있는 경우
• TOOL_SOURCE가 local·mcp·both 밖의 값인 경우
• MCP_TRANSPORT가 streamable_http·sse 밖의 값인 경우 (TOOL_SOURCE가 mcp·both일 때)

TOOL_SOURCE 오타는 특히 조용합니다. 검증이 없으면 도구 공급원 0개로 멀쩡히 뜬 다음, 모든 요청이 근거 가드에 걸려 "도구가 전부 실패한 것"처럼 보이게 됩니다.

검사는 이렇게 돌립니다.

pytest
ruff check .
mypy aiagent app main.py
pytest --cov            # 커버리지가 필요할 때

호출 예시

# 단건 응답
curl -X POST http://localhost:8000/api/v1/chat \
  -H 'Content-Type: application/json' \
  -d '{"message": "demo-model로 hello 추론해줘"}'

# SSE 스트리밍 (session → message → analysis → done)
curl -N -X POST http://localhost:8000/api/v1/chat-stream \
  -H 'Content-Type: application/json' \
  -d '{"message": "demo-model로 hello 추론해줘", "conversation_id": "conv-1"}'

응답 헤더에 X-Request-ID가 붙어 나옵니다. 서버 로그 앞머리의 [request_id]와 같은 값이라, "어제 그 답변 좀 이상했어요" 같은 문의를 받아도 이 값 하나로 해당 요청만 골라낼 수 있습니다.

도구는 어디서 오는가

로컬 구현일 수도 있고, MCP 서버가 들고 있을 수도 있습니다. 둘 다 쓸 수도 있고요. 오케스트레이터는 이 차이를 모릅니다. TOOL_SOURCE 값 하나로 정해집니다.

값	주입되는 provider	쓰임새
local	LocalToolProvider	도구를 이 저장소가 소유할 때
mcp	McpToolProvider	도구를 외부 MCP 서버가 소유할 때
both	둘 다	공용 도구는 MCP로, 서비스 전용 도구는 로컬로

MCP 쪽은 langchain-mcp-adapters가 필요한데, 선택 의존성으로 빼고 import를 지연시켰습니다. 설치하지 않아도 local 모드는 멀쩡히 뜹니다. 두 공급원에서 같은 이름의 도구가 오면 먼저 로드된 쪽을 쓰고 경고를 남깁니다.

MCP_TIMEOUT_SECONDS는 전송 방식에 따라 표현이 다릅니다. streamable_http는 timedelta를, sse는 초 단위 실수를 받습니다. 어댑터가 이 변환을 맡습니다.

교체 지점

실제 서비스로 만들 때 손댈 곳은 여기가 전부입니다.

무엇	어디	어떻게
외부 모델 API 호출	app/adapters/http_inference_client.py	실제 API 스키마에 맞춰 _build_payload·_parse_response 수정
도구 추가	aiagent/tools/tool_registry.py + tool_contract.py	포트를 주입받아 build_tools에 등록하고, 이름·시나리오·반환 모델을 계약에 등록
도구 반환 계약	aiagent/tools/tool_contract.py	DICT_TOOL_RETURN_MODEL에 도구별 Pydantic 모델 등록 (등록하지 않으면 검증 없이 통과)
의도 종류	aiagent/constants.py + tool_contract.py	IntentType에 추가하고 시나리오·근거필수 목록에 매핑
도구 공급원	.env의 TOOL_SOURCE + main.py:build_tool_providers	새 공급원은 ToolProvider를 구현해 주입
대화 이력 백엔드	app/agent/conversation_store.py	ConversationStore를 DB나 Redis 구현으로 교체
실패 안내 문구	aiagent/constants.py	DICT_ERROR_KIND_USER_MESSAGE의 원인별 문구를 서비스 말투로 교체
근거 길이 상한	aiagent/agents/agent_orchestrator.py	쓰는 모델의 입력 한도에 맞춰 MAX_EVIDENCE_CHARS 조정

도구 하나 추가하는 데 세 군데를 건드려야 하는 게 번거로울 수 있습니다. 반환 계약을 강제하려다 보니 이렇게 됐는데, 빠뜨려도 그 도구만 검증을 건너뛸 뿐 나머지는 멀쩡히 돕니다.

실패했을 때 어떻게 되나

이 골격에서 제일 신경 쓴 부분입니다.

LLM은 근거가 비어 있어도 답을 만들어냅니다. 그래서 도구가 전부 실패한 요청과 제대로 조회해서 분석한 요청이 사용자 눈에는 똑같이 보입니다. 숫자까지 그럴듯하게 붙어서요. 프롬프트로 막으려 들면 결국 모델이 지시를 지켜주기를 바라는 셈이 됩니다.

그래서 그래프 라우팅으로 처리했습니다. 성공한 도구 결과가 하나도 없으면 합성 노드를 아예 실행하지 않고, 실패 사유가 담긴 확정 문구를 그대로 답변으로 씁니다. 전량 실패가 일시적인 장애일 수도 있으니 MAX_TOOL_RETRIES까지는 계획부터 다시 시도합니다.

여기 딸려 오는 규칙이 몇 가지 있습니다.

• 도구 응답은 DICT_TOOL_RETURN_MODEL의 계약으로 검증합니다. 계약을 벗어나면 그 호출은 실패입니다. 어긋난 필드 경로는 str_error에 그대로 남습니다. str_output: Field required 같은 식으로요. 다만 이 문장은 로그와 진단 응답에만 남고 답변에는 쓰지 않습니다.
• 사용자에게 나가는 문구는 따로 만듭니다. 우리가 쓴 안내가 있으면 str_user_message를, 없으면 str_error_kind별 기본 문구를 씁니다. 예외 원문은 사용자가 읽을 글이 아닙니다.
• 실패에는 원인 구분이 붙습니다. 도구 제공 측(contract_violation·transport·tool_error)과 에이전트 측(not_implemented·unavailable·internal)을 갈라, 장애 문의가 왔을 때 어느 쪽을 봐야 하는지 바로 나오게 했습니다.
• 의도에 없는 도구를 계획했다면 조용히 버리지 않고 실패로 남깁니다. 실행 0건과 성공 0건은 전혀 다른 상황인데, 버려 버리면 둘을 구분할 수 없게 됩니다.
• LLM 호출 자체가 죽는 경우도 같이 다룹니다. 의도 분류가 실패하면 기본 의도로 넘어가고, 계획이나 합성이 실패하면 장애 안내 문구로 마무리합니다. 모델 API가 내려가도 500 대신 답변이 나갑니다.
• NotImplementedError는 RuntimeError의 하위 타입이라 실패 예외 목록에 이미 걸립니다. 미구현 도구가 조용히 통과하면 근거 판정 자체가 무의미해지니 이게 맞습니다. 원인 구분에서는 RuntimeError보다 먼저 확인해야 internal로 묻히지 않습니다.
• 답변 프롬프트에 넣는 근거에는 MAX_EVIDENCE_CHARS 상한이 있습니다. 목록을 돌려주는 도구가 붙으면 근거가 모델 입력 한도를 넘어 답변 대신 400이 돌아옵니다. 상한에서 자르고 잘렸다는 사실을 근거 끝에 남겨, 남은 앞부분만 보고 "전부 조회했다"는 투로 답하지 않게 합니다.

이렇게 정한 배경은 docs/adr/에 남겨 뒀습니다.

아직 안 된 것들

골격이라 일부러 남겨 둔 것들입니다. 이대로 운영에 올리면 안 됩니다.

대화 이력은 임시 구현입니다

InMemoryConversationStore는 붙일 백엔드가 정해지기 전까지 동작만 시켜 두려고 만든 것입니다. 실제 저장소는 스키마·보존기간·격리 정책이 얽히는 영역이라 백엔드·DB 담당과 함께 설계해야 합니다. 그때까지 남아 있는 문제는 셋입니다.

• 프로세스 메모리에 있습니다. 워커를 두 개 이상 띄우거나 재시작하면 이력이 끊깁니다.
• 동시 요청 시 한 턴이 사라집니다. _run_chat은 이력을 읽고 → 에이전트를 실행하고 → 통째로 덮어쓰는데, 그 사이에 락이 없습니다. 같은 conversation_id로 두 요청이 겹치면 나중에 저장한 쪽이 이기고 먼저 끝난 턴은 이력에서 사라집니다. 답변은 이미 사용자에게 나간 뒤라, 다음 턴에서 모델이 방금 한 말을 기억하지 못하는 형태로 드러납니다. 재시작이나 다중 워커와 무관하게 단일 워커에서도 재현됩니다. 프로세스 안에서 락을 걸어도 워커가 둘 이상이면 다시 깨지므로, 저장소가 "이력 통째로 덮어쓰기" 대신 "턴 덧붙이기"를 원자적으로 제공하는 쪽이 맞습니다. 백엔드를 고를 때 같이 정할 일입니다.
• ChatRequest.user_id를 받기만 하고 쓰지 않습니다. 이력이 conversation_id만으로 키잉되어, 사용자별로 격리되는 것처럼 보이지만 실제로는 격리되지 않습니다.

그 밖에

인증도 없습니다. 엔드포인트가 그냥 열려 있으니 API 키든 JWT든 앞단에 두어야 합니다.

나머지는 목록으로 정리합니다.

• 재시도는 전량 실패에 대해 1회 고정입니다. 지수 백오프나 도구별 정책은 없습니다.
• HTTP 어댑터에 재시도나 서킷브레이커가 없습니다. 타임아웃만 걸려 있습니다.
• SSE는 단계 단위로 흘려보냅니다. 토큰 단위로 스트리밍하려면 합성 노드에서 astream을 써야 합니다.
• 토큰이나 비용을 집계하지 않습니다. 필요하면 콜백을 붙이면 됩니다.
• 프롬프트 인젝션 방어가 없습니다. 도구 인자를 LLM이 채우는 구조라, 파일을 지운다든가 하는 파괴적인 도구를 붙일 거면 인자 검증과 승인 절차를 따로 설계해야 합니다.

핸드오프 절차는 docs/INTEGRATION_GUIDE.md에 정리해 뒀습니다.