원문은 AIdeazz에 처음 게시되었으며, 정식 링크와 함께 이곳에 교차 게시되었습니다.

텔레그램 봇을 위한 간단한 작업 라우터였던 저의 첫 LangGraph 에이전트는 2주 동안 작업의 100%를 소리 없이 놓쳤습니다. StateSchema는 TypedDict로 정의되어 있었지만, 저는 Pydantic 모델 인스턴스를 전달하고 있었습니다. LangGraph의 MemorySaver는 오류를 로그에 남기지 않고 단순히 이 타입 불일치(type mismatch)를 삼켜버렸습니다. 에이전트는 실행되는 것처럼 보였지만, 상태(state)는 전혀 유지되지 않았습니다. 다음 노드에서 작업을 전혀 가져오지 못했습니다. 이는 제가 Oracle Cloud에서 안정적인 상태 유지(stateful) LangGraph 에이전트를 프로덕션에 배포하기 전 수행한 세 번의 전체 재작성 중 첫 번째였습니다.

소리 없는 StateSchema 불일치

초기 에이전트는 텔레그램으로부터 들어오는 사용자 요청을 받아 분류하고, 특화된 LLM 에이전트(빠르고 간단한 쿼리를 위한 Groq, 복잡한 멀티턴 작업을 위한 Claude 3 Opus)로 라우팅하도록 설계되었습니다. 핵심 아이디어는 상태(state)를 위한 TypedDict를 사용하는 것이었습니다:

class AgentState(TypedDict):
    chat_id: int
    user_input: str
...

새 메시지가 도착하면 이 상태를 초기화하여 그래프에 전달했습니다. 문제는 UserSession이나 JobDetails와 같이 더 복잡한 객체를 나타내는 Pydantic 모델로 상태를 풍부하게 만들려고 할 때 발생했습니다. AgentState['user_input'] = job_details.description과 같이 업데이트하는 대신, 저는 str을 기대하는 필드에 JobDetails Pydantic 객체를 직접 저장하려고 시도했습니다.

LangGraph의 MemorySaver(그리고 이후에 시도했던 SqliteSaver)는 얕은 복사(shallow copy) 및 업데이트를 수행합니다. 만약 들어오는 상태 딕셔너리(state dict)에 TypedDict 스키마와 일치하지 않는 키가 있거나, 일치하는 키의 값의 *타입(types)*이 일치하지 않더라도 에러를 발생시키지 않습니다. 단지 일치하지 않는 필드에 대한 업데이트를 무시할 뿐입니다. 저의 job_details Pydantic 객체는 user_input에 할당되었음에도 불구하고 실제로 저장되지 않았습니다. user_input 필드는 None이거나 초기 값으로 남아 있었고, 결과적으로 라우팅(routing)에 필요한 핵심 정보가 폐기되었습니다.

해결책은 LangGraph 상태를 업데이트하기 전에 Pydantic 모델을 딕셔너리(dictionary)나 JSON 문자열로 엄격하게 직렬화(serialize)하는 것이었습니다. 예를 들어:

# 잘못된 방법:
# state['job_details'] = job_details_pydantic_instance

...

이로 인해 저는 AgentState를 임의의 Python 객체를 담는 유연한 컨테이너가 아니라, 엄격한 데이터 전송 객체(Data Transfer Object, DTO)로 취급해야만 했습니다.

체크포인트 손상과 Oracle Object Storage

상태 스키마를 수정한 후, 저는 더 견고한 체크포인팅(checkpointing) 메커니즘으로 넘어갔습니다. SqliteSaver는 로컬 개발에는 괜찮지만, Oracle Cloud Infrastructure (OCI) Container Instances에서 실행되는 프로덕션 에이전트에는 적합하지 않습니다. 제 에이전트들은 컴퓨팅 측면에서는 상태가 없지만(stateless), 그래프 실행 측면에서는 상태를 유지(stateful)합니다. 저에게는 외부 지속성(external persistence)이 필요했습니다.

처음에는 OCI Object Storage를 직접 사용하는 것을 시도했습니다. LangGraph의 MemorySaver는 확장 가능하며, 저는 상태를 JSON으로 직렬화하여 OCI Object Storage의 S3 호환 버킷(S3-compatible bucket)에 업로드하는 커스텀 OCISaver를 작성했습니다.

문제점: 동시 업데이트(concurrent updates). LangGraph의 기본 MemorySaver(및 그 파생물들)는 단일 작성자를 가정하거나 로컬 파일에 대해 내부적으로 동시성을 처리합니다. 제 에이전트의 여러 인스턴스(예: 서로 다른 chat_id를 처리하는 경우)가 Object Storage의 각 체크포인트를 업데이트하려고 시도할 때, 레이스 컨디션(race conditions) 문제에 직면했습니다. 에이전트가 상태를 읽고, 이를 처리한 다음, 다시 쓰려고 시도하는 과정에서, 그 사이에 다른 에이전트가 업데이트를 작성했다면 첫 번째 에이전트의 쓰기 작업이 더 최신인 상태를 덮어쓰게 되어, 대화 턴(turns)이 유실되거나 히스토리가 손상되는 결과로 이어졌습니다.

S3와 마찬가지로 OCI Object Storage는 최종 일관성(eventual consistency) 모델을 따릅니다. 새로운 객체에 대해서는 강력한 쓰기 후 읽기 일관성(read-after-write consistency)을 제공하지만, 덮어쓰기(overwrites)의 경우 최종 일관성을 보일 수 있습니다. 더 결정적으로, 객체 콘텐츠에 대해 직접적인 원자적 비교 및 교체(atomic compare-and-swap) 작업을 제공하지 않습니다. 객체 버전 관리(object versioning)를 사용할 수는 있지만, 이는 단지 새로운 버전을 생성할 뿐이며, 추가적인 로직 없이는 덮어쓰기를 방지하거나 낙관적 잠금(optimistic locking) 메커니즘을 제공하지 않습니다.

제가 작성한 커스텀 OCISaver는 다음과 같은 모습이었습니다:

class OCISaver(BaseCheckpointSaver):
    def __init__(self, bucket_name: str, namespace: str, object_storage_client):
        self.bucket_name = bucket_name
...

put_tuple 메서드가 원인이었습니다. 그것은 맹목적인 쓰기(blind write)였습니다.

해결책은 체크포인팅을 위한 적절한 데이터베이스를 도입하는 것이었습니다. 저는 Oracle Autonomous Database (ADB) Serverless, 특히 그 JSON Document Store 기능을 선택했습니다. 각 thread_id(사용자의 채팅 세션에 매핑됨)는 하나의 문서(document)가 되었습니다. ADB는 트랜잭션 일관성(transactional consistency)을 제공합니다. 저는 트랜잭션 내에서 읽기-수정-쓰기(read-modify-write) 작업을 수행할 수 있었고, 이를 통해 두 에이전트가 동일한 thread_id의 체크포인트를 업데이트하려고 시도할 경우 하나는 성공하고 다른 하나는 대기하거나 실패하게 하여 재시도 로직(retry logic)을 적용할 수 있도록 보장했습니다.

이를 위해서는 버전 번호를 확인하는 WHERE 절과 함께 SQL UPDATE 문을 사용하거나, ADB의 네이티브 JSON 업데이트 기능을 활용하도록 BaseCheckpointSaver 구현을 완전히 재작성해야 했습니다.

# ADB를 위한 단순화된 개념적 예시
class ADBSaver(BaseCheckpointSaver):
    def __init__(self, db_connection_pool):
...

버전 컬럼을 사용하여 낙관적 잠금 (Optimistic Locking)을 사용하는 이 패턴은 분산 시스템 내의 모든 공유 가능한 가변 상태 (Shared Mutable State)에 있어 매우 중요합니다.

단 하나의 패턴: 명시적 상태 전이와 에이전트 오케스트레이션 (Agent Orchestration)

강력한 체크포인팅 (Checkpointing) 기능이 있음에도 불구하고, 저의 멀티 에이전트 시스템은 여전히 취약했습니다. 핵심 문제는 암시적 상태 전이 (Implicit State Transitions)와 긴밀하게 결합된 에이전트 로직이었습니다. 에이전트가 다음 단계를 결정하면 그래프는 단순히 그것을 실행할 뿐이었습니다. 만약 에이전트가 잘못된 결정(예: 쿼리를 잘못 분류하거나 루프에 빠지는 경우)을 내리면, 전체 그래프가 그대로 따라가게 됩니다.

저의 에이전트들은 다음과 같았습니다:

1. 라우터 에이전트 (Router Agent): 사용자 입력을 분류합니다 (Groq).
2. 검색 에이전트 (Search Agent): 필요 시 웹 검색을 수행합니다 (Claude 3 Haiku).
3. 요약 에이전트 (Summarizer Agent): 검색 결과를 요약합니다 (Groq).
4. 응답 에이전트 (Response Agent): 최종 응답을 생성합니다 (Claude 3 Opus).

초기 그래프는 상태(State) 내의 classification 필드에 기반한 일련의 조건부 엣지 (Conditional Edges)로 구성되었습니다. 만약 classification == "search"라면 search_agent로 이동합니다. 만약 search_agent가 실패하면 단순히 빈 결과를 반환하게 되고, summarizer_agent는 아무것도 받지 못해 결국 품질이 낮은 최종 응답으로 이어지게 됩니다. 그래프 구조 자체 내에 명시적인 에러 핸들링 (Error Handling)이나 재시도 메커니즘 (Retry Mechanism)이 없었습니다.

세 번째 재작성에서는 더욱 명시적인 오케스트레이션 패턴을 도입했습니다:

• 상태 기반 결정 (State-driven decisions): 모든 노드의 출력은 next_step 필드를 포함하여 상태 (State)를 명시적으로 업데이트합니다.
• 감독 에이전트 (Supervisor Agent): (속도를 위해 Groq를 사용하는) 전용 LLM이 메타 에이전트 (Meta-agent) 역할을 수행하며, 현재 상태를 검토하고 다음에 실행할 노드를 결정합니다. 이 에이전트는 항상 조건부 엣지 (Conditional edge)의 대상이 됩니다.
• 도구형 에이전트 (Tool-like Agents): 각각의 특화된 에이전트 (Router, Search, Summarizer, Response)는 도구 (Tool)와 더 유사하게 취급됩니다. 에이전트는 현재 상태를 받아 특정 작업을 수행하고, 그 *결과 (Result)*와 제안된 next_step을 감독 에이전트에게 반환합니다.

AgentState는 다음과 같이 확장되었습니다:

class AgentState(TypedDict):
    chat_id: int
    user_input: str
...

그래프 구조는 다음과 같이 변경되었습니다:

graph = StateGraph(AgentState)

graph.add_node("router", router_agent)
...

supervisor_agent의 프롬프트 (Prompt)가 매우 중요합니다. 이 에이전트는 전체 AgentState를 전달받으며, 미리 정의된 next_step 값 중 하나를 출력하도록 지시받습니다. 예를 들어:

당신은 AI 오케스트레이터 (Orchestrator)입니다. 대화의 현재 상태와 마지막 작업의 출력을 검토하십시오.
다음 논리적 단계를 결정하십시오.
현재 상태: {state}
...

이 패턴은 그래프를 훨씬 더 견고하게 만들었습니다. 감독 에이전트는 다음과 같은 작업을 수행할 수 있습니다:

• 재시도 (Retry): 만약 search_agent가 실패하면, 감독 에이전트는 검색을 재시도하거나 심지어 다른 검색 전략으로 경로를 재설정 (Re-route)하도록 결정할 수 있습니다.
• 교정 (Correct): 만약 router_agent가 잘못 분류했다면, 감독 에이전트는 정제된 프롬프트와 함께 다시 router_agent로 보내거나 분류를 직접 무효화 (Override)할 수 있습니다.
• 우아한 에러 처리 (Handle Errors Gracefully): 감독 에이전트에 의해 전용 error_handler_node가 호출될 수 있으며, 이 노드는 에러를 기록하거나, 사용자에게 알리거나, 폴백 (Fallback)을 시도할 수 있습니다.

이러한 명시적인 오케스트레이션은 트랜잭션 체크포인팅 (Transactional checkpointing) 및 엄격한 상태 스키마 (State schema) 준수와 결합되어, 마침내 안정적인 프로덕션 환경용 멀티 에이전트 시스템 (Multi-agent system)으로 이어졌습니다. 코드는 더 장황해졌지만, 제어 흐름 (Control flow)과 에러 처리의 명확성은 실제 운영 중인 에이전트를 디버깅할 때 매우 귀중한 가치를 제공합니다.

자주 묻는 질문 (Frequently Asked Questions)

Q: 감독 에이전트 (Supervisor Agent) 대신 오케스트레이션 (Orchestration)을 위해 왜 하나의 거대한 LLM을 사용하지 않나요?
A: 오케스트레이션 (Orchestration)을 위해 전용의 더 작은 LLM (Groq의 Llama 3 8B와 같은)을 사용하는 것이 거대 모델보다 더 빠르고 저렴합니다. 이는 오직 제어 흐름 (Control Flow) 결정에만 집중하여 파이프라인의 _로직 (Logic)_에 대한 환각 (Hallucination) 가능성을 줄이는 한편, 더 큰 모델 (Claude 3 Opus)은 복잡한 콘텐츠 생성 (Content Generation)을 처리하도록 합니다.

Q: OCI에서 체크포인팅 (Checkpointing)을 위해 구체적으로 어떤 데이터베이스를 사용했나요?
A: JSON Document Store로 구성된 Oracle Autonomous Database (ADB) Serverless를 사용했습니다. 이는 ACID 트랜잭션을 제공하고 자동으로 확장되는데, 이는 예측 불가능한 에이전트 부하 (Agent Load)에 있어 매우 중요합니다.

Q: 프로덕션 환경에서 AgentState의 스키마 진화 (Schema Evolution)를 어떻게 처리하나요?
A: 내부적으로 AgentState에 Pydantic을 사용하여 기본값 (Default values)과 선택적 필드 (Optional fields)를 허용합니다. 새 버전을 배포할 때는 새로운 필드를 선택적 또는 기본값과 함께 추가함으로써 하위 호환성 (Backward compatibility)을 보장합니다. 중대한 변경 사항 (Breaking changes)이 있는 경우에는 ADB에 있는 기존 체크포인트 문서들을 업데이트하기 위한 마이그레이션 스크립트 (Migration script)가 필요합니다.

Q: 감독 에이전트 (Supervisor Agent)의 전형적인 지연 시간 (Latency) 오버헤드는 어느 정도인가요?
A: Groq를 사용할 경우, 감독 에이전트는 결정당 약 100-200ms의 지연 시간을 추가하며, 이는 대부분의 대화형 에이전트 (Conversational agents)에게 수용 가능한 수준입니다. 이는 에이전트가 루프 (Loops)에 빠지거나 조용히 실패할 가능성과 비교했을 때, 안정성과 디버깅 가능성 (Debuggability)을 높이기 위해 지불할 만한 작은 비용입니다.

Q: 이 설정에서 에이전트의 성능을 어떻게 모니터링하고 문제를 식별하나요?
A: 각 노드 (Node)는 chat_id 및 current_task와 함께 진입 (Entry), 퇴장 (Exit), 그리고 모든 오류를 로그로 기록합니다. 이러한 로그를 OCI Logging Analytics로 전송합니다. supervisor_agent의 결정 또한 로그로 기록되어 에이전트의 사고 과정 (Thought process)과 제어 흐름 (Control flow)에 대한 명확한 추적 (Trace)을 제공하며, 이를 통해 에이전트가 어디서 경로를 벗어났는지 정확히 찾아내기가 더 쉬워집니다.

— Elena Revicheva · AIdeazz · Portfolio