AI 서비스 기초 코드 & 에이전트 워크플로 구축

[VitoJeong]

개요

AI 서비스의 핵심인 멀티에이전트 오케스트레이션 파이프라인과 semantic 검색 체계를 구축했습니다.

구현 항목

작업내용 상세 문서는 on-seoul-agent/docs/ai-service-implementation.md를 참고해주세요.

다중 DB 연동 및 임베딩 인프라 (Phase 4, 7)

• DB 계정 분리: on_ai_app(CRUD) / on_data_reader(SELECT 전용) 이중 엔진 구성 및 FastAPI DI 주입
• 권한 격리 검증: on_data_reader의 INSERT/UPDATE 시도 시 권한 오류 발생 통합 테스트
• 임베딩 배치 적재: embed_metadata.py — --all / --limit / --incremental 모드 지원, 신규 service_id만 재임베딩하는 증분 업데이트

LangChain Agent 구현 (Phase 5)

• RouterAgent: LCEL 체인으로 사용자 의도를 SQL_SEARCH / VECTOR_SEARCH / MAP / FALLBACK 4종으로 분류
• SqlAgent: LLM이 추출한 파라미터(카테고리·지역·키워드·상태)를 바인드 파라미터 SQL로 변환 — LLM 생성 SQL 직접 실행 없음
• VectorAgent: 질의 정제 → 임베딩 → tools/vector_search.py 위임
• AnswerAgent: 검색 결과 통합 → 자연어 답변 생성, 첫 메시지 시 대화 제목 생성(title_needed 플래그)

워크플로우 및 Trace (Phase 6)

• AgentWorkflow: Router → Branch(SQL/Vector/MAP/FALLBACK) → Answer 순서 실행, 세션 주입 방식으로 테스트 용이성 확보
• DB 세션 라우팅: SqlAgent → on_data(읽기 전용), VectorAgent → on_ai(service_embeddings)
• chat_agent_traces 적재: 실행 완료 후 intent·node_path·elapsed_ms를 JSONB로 저장 (best-effort, 실패 시 응답 영향 없음)
• 오류 복원: Agent 예외 발생 시 fallback 답변 반환, state.error 기록

검색 전략 및 파라미터 튜닝 (Phase 8, 9)

• tools/vector_search.py: pre-filter(카테고리·지역·상태를 metadata JSONB WHERE 절 적용) + pgvector 코사인 유사도 검색 독립 함수
• HNSW 인덱스: m=16, ef_construction=64, ef_search=40 DDL 활성화 (1000건 기준)
• 전략 결정: 하이브리드 FTS·재순위화 미채택 — 소규모 데이터에서 순수 벡터 검색 품질 충분

검색 품질 평가 인프라 (Phase 10)

• eval_search.py: 카테고리·지역·키워드·의미별 샘플 쿼리 20건, --query / --top-k 옵션 지원

[f-lab-ted]

ai-service-implementation.md Phase 4~10에 명시된 멀티에이전트 오케스트레이션 파이프라인과 semantic 검색 체계를 리뷰했습니다. Router → Branch(SQL/Vector/Map/Fallback) → Answer 워크플로우 구조, 이중 DB 세션 라우팅(on_ai CRUD / on_data SELECT 전용), 임베딩 배치 적재 스크립트, pgvector 검색 도구, 그리고 13건의 워크플로우 통합 테스트를 중점적으로 검토했습니다. 특히 PR 가이드라인에 따라 SOLID 설계 원칙 준수 여부, 테스트 적절성, LLM 호출 안정성에 초점을 맞추어 리뷰했습니다.

잘한 점

SqlAgent에서 LLM이 SQL을 직접 생성하지 않고 파라미터만 추출하여 바인드 파라미터로 주입하는 설계는 SQL Injection 방지 측면에서 훌륭합니다. AgentState TypedDict 기반의 불변 스프레드 패턴({**state, key: value})과 생성자 주입 방식으로 테스트 용이성을 확보한 점, 그리고 vector_search.py의 화이트리스트 기반 pre-filter 조건 조립도 보안과 확장성 모두 잘 고려되어 있습니다.

보완할 점

가장 우려되는 부분은 LLM Chat 호출의 안정성입니다. 임베딩 API에는 rate limit + 429 백오프가 잘 구현되어 있지만, 핵심 Chat LLM 호출(Router/SQL/Vector/Answer 4개 에이전트)에는 타임아웃이나 재시도가 전혀 없어 프로덕션에서 LLM 응답 지연 시 요청이 무한 대기할 수 있습니다. _save_trace의 세션 롤백 누락과 429 판별의 문자열 매칭 방식도 안정성 관점에서 개선이 필요합니다.

결론

전체적으로 설계 문서대로 잘 구현되어 있고 테스트도 충실합니다. 다만 LLM Chat 호출에 타임아웃/재시도 메커니즘을 추가하는 것은 프로덕션 안정성을 위해 반드시 수정 바랍니다. 나머지 코멘트는 검토 후 반영 여부 결정해주세요.

[f-lab-ted]

[Critical — LLM Chat 호출 타임아웃/재시도/Rate Limit 대책 부재]

임베딩 API에는 AsyncLimiter 기반 RPM 제한 + 429 지수 백오프(최대 5회, 10s→160s)가 잘 구현되어 있지만, 핵심 워크플로우의 4개 에이전트(RouterAgent.classify, SqlAgent.search, VectorAgent.search, AnswerAgent.answer)가 내부에서 호출하는 self._chain.ainvoke()에는 타임아웃, 재시도, Rate Limit 제어가 모두 없습니다.

1. 타임아웃 부재: LLM 프로바이더가 일시적으로 느려지거나 응답하지 않으면 워크플로우 전체가 무한 대기합니다. FastAPI 워커 스레드를 점유하게 되어 서비스 전체에 영향을 줄 수 있습니다.
2. Rate Limit 미처리: llm/client.py에서 ChatOpenAI / ChatGoogleGenerativeAI 생성 시 max_retries나 rate limit 관련 설정이 없습니다. 임베딩 쪽에는 AsyncLimiter와 429 재시도가 있는 것과 대조적으로, Chat LLM 호출은 프로바이더의 rate limit에 무방비 상태입니다. 동시 요청이 몰리면 429 응답을 받고 그대로 예외가 전파됩니다.

제안:

# 방법 1: asyncio.wait_for로 타임아웃 + LangChain with_retry()로 재시도
import asyncio

self._chain = (
    prompt
    | llm.with_structured_output(_IntentOutput)
    .with_retry(stop_after_attempt=3, wait_exponential_jitter=True)
)

async def classify(self, state: AgentState) -> AgentState:
    result = await asyncio.wait_for(
        self._chain.ainvoke({"message": state["message"]}),
        timeout=30.0,  # 30초
    )
    return {**state, "intent": result.intent}

# 방법 2: LLM 클라이언트 레벨에서 max_retries 설정
ChatOpenAI(model=..., max_retries=3, request_timeout=30)

# 방법 3: 임베딩과 동일하게 AsyncLimiter 적용
from aiolimiter import AsyncLimiter
_chat_limiter = AsyncLimiter(max_rate=10, time_period=60)

async def classify(self, state: AgentState) -> AgentState:
    async with _chat_limiter:
        result = await asyncio.wait_for(
            self._chain.ainvoke({"message": state["message"]}),
            timeout=30.0,
        )
    return {**state, "intent": result.intent}

임베딩 호출에 적용된 것과 동일한 수준의 방어(AsyncLimiter + 지수 백오프 재시도)를 Chat LLM 호출에도 적용하는 것을 권장합니다. 최소한 asyncio.wait_for 타임아웃과 with_retry()만이라도 적용하면 무한 대기 및 429 폭주를 방지할 수 있습니다.

[f-lab-ted]

[Major — _save_trace 세션 롤백 누락]

_save_trace는 finally 블록에서 호출되므로, 메인 워크플로우에서 ai_session을 사용하던 중(VectorAgent) 예외가 발생하면 세션의 트랜잭션이 failed 상태일 수 있습니다. 이 상태에서 rollback 없이 바로 INSERT를 시도하면 InFailedSqlTransaction 오류가 발생하고, except에서 잡히긴 하지만 trace가 유실됩니다.

또한 execute 성공 후 commit 실패 시에도 rollback이 없어 세션이 dirty 상태로 남습니다.

async def _save_trace(session, message_id, trace):
    try:
        await session.rollback()  # 이전 트랜잭션 상태 정리
        trace_json = json.dumps(trace, ensure_ascii=False, default=str)
        await session.execute(
            text("INSERT INTO chat_agent_traces ..."),
            {"message_id": message_id, "trace": trace_json},
        )
        await session.commit()
    except Exception as exc:
        logger.warning("trace 저장 실패 (message_id=%s): %s", message_id, exc)
        try:
            await session.rollback()
        except Exception:
            pass

best-effort 의도에 맞게 rollback을 선행하면 trace 저장 성공률이 높아집니다.

[f-lab-ted]

[Major — 429 판별 문자열 매칭의 취약성]

is_rate_limit = "429" in str(exc) or "RESOURCE_EXHAUSTED" in str(exc)

이 방식은 다음 문제가 있습니다:

1. 예외 메시지 포맷이 라이브러리 버전에 따라 변경될 수 있음
2. 우연히 "429"가 포함된 비관련 에러도 재시도 대상이 됨
3. google.api_core.exceptions.ResourceExhausted 등 구체적인 예외 타입이 있는데 활용하지 않음

제안:

from google.api_core.exceptions import ResourceExhausted
from openai import RateLimitError

is_rate_limit = isinstance(exc, (ResourceExhausted, RateLimitError))
# 혹은 HTTP status code 기반 판별
if hasattr(exc, 'code'):
    is_rate_limit = exc.code == 429

예외 타입 기반 판별이 더 안정적이고, 향후 OpenAI 임베딩 지원 시에도 확장이 용이합니다.

[f-lab-ted]

[Minor — embed_documents / embed_query 동기 메서드에 rate limit 미적용]

_GeminiEmbeddings의 비동기 메서드(aembed_query, aembed_documents)에는 rate limit + 429 재시도가 적용되어 있지만, 동기 메서드(embed_documents, embed_query)는 self._base에 그대로 위임하여 rate limit이 우회됩니다.

def embed_documents(self, texts: list[str]) -> list[list[float]]:
    return self._base.embed_documents(texts)  # rate limit 없음

def embed_query(self, text: str) -> list[float]:
    return self._base.embed_query(text)  # rate limit 없음

현재 코드에서 동기 메서드를 직접 호출하는 곳이 없다면 당장은 문제가 없지만, 의도치 않은 호출 시 rate limit 없이 API를 때릴 수 있습니다. 동기 메서드에서 raise NotImplementedError("Use async methods")를 던지거나, Embeddings 인터페이스 구현 의무상 필요하다면 docstring에 rate limit 미적용을 명시하는 것이 안전합니다.

[f-lab-ted]

[Minor — 예외 시 엔진 dispose 누락]

run() 함수에서 임베딩 처리 중 예외가 발생하면 on_data_engine.dispose()와 on_ai_engine.dispose()가 호출되지 않아 커넥션 풀이 정리되지 않습니다. 배치 스크립트라서 프로세스 종료 시 OS가 정리하긴 하지만, try/finally로 감싸는 것이 좋습니다.

async def run(limit, incremental=False):
    on_data_engine = create_async_engine(...)
    on_ai_engine = create_async_engine(...)
    try:
        # ... 기존 로직 ...
    finally:
        await on_data_engine.dispose()
        await on_ai_engine.dispose()

[f-lab-ted]

[Minor — "프로파이더" 오타]

"프로파이더"는 "프로바이더(provider)"의 오타입니다. 이 PR에서 "벤더 → 프로바이더" 용어 통일 작업이 있었는데, exceptions.py, ai-service-implementation.md 등 여러 곳에서 "프로파이더"로 잘못 기재되어 있습니다.

# before
"""LLM 프로파이더(Gemini, OpenAI 등) 호출 관련 예외"""

# after
"""LLM 프로바이더(Gemini, OpenAI 등) 호출 관련 예외"""

[VitoJeong]

@f-lab-ted 피드백 주신 내용 반영하여 merge 하겠습니다~!