ContextVar로 Multi-Agent 상태관리 하기

ContextVar로 Multi-Agent 상태관리 하기

Multi-Agent 시스템에서 가장 까다로운 문제 중 하나가 Agent 간 상태 공유다. 특히 파킹통장 추천 AI 프로젝트에서는 QuestionAgent 내부의 5개 Tool이 순차적으로 실행되면서 특정 데이터에 선택적으로 접근해야 하는 복잡한 상황이 발생했다.

핵심 문제는 이렇다.

QuestionAgent Tool 체인

🔄 Tool 실행 순서 및 역할

1. ConditionExtractorTool

• 역할: 우대조건 및 금리정보 청크 데이터 추출
• 입력: EligibilityAgent에서 전달받은 적격 통장 목록
• 처리: MongoDB에서 각 통장의 우대조건 상세 정보 조회
• 출력: 구조화된 청크 데이터 (basic_rate_info, preferential_details)

2. PatternAnalyzerTool

• 역할: LLM 기반 우대조건 패턴 분석 및 RAG 쿼리 생성
• 입력: 추출된 청크 데이터
• 처리: 우대조건 텍스트를 분석하여 공통 패턴 식별
• 출력: 표준화된 패턴 목록 + RAG 검색용 쿼리

3. QuestionGeneratorTool

• 역할: 패턴 분석 결과 기반으로 RAG 검색하여 사용자 질문 생성
• 입력: 분석된 패턴과 RAG 쿼리
• 처리: RAG 시스템에서 관련 정보 검색 후 Yes/No 질문 생성
• 출력: 카테고리별 사용자 질문 리스트

4. UserInputTool

• 역할: 환경별 적응형 사용자 입력 처리 (콘솔/API 전환)
• 입력: 생성된 사용자 질문 리스트
• 처리: 테스트 모드에 따라 콘솔 입력 또는 API 응답 처리
• 출력: 사용자 응답 데이터 (질문별 Yes/No 답변)

5. ResponseFormatterTool

• 역할: QuestionAgent 최종 출력 포맷팅 (StrategyAgent 입력용)
• 입력: 사용자 응답 데이터
• 처리: Context에서 eligible_products와 user_conditions 조회
• 출력: QuestionSuccessResponse (다음 Agent로 전달할 표준 포맷)

---

🎯 핵심 문제점

Context 접근이 필요한 Tool

• ConditionExtractorTool: eligible_products 필요
• ResponseFormatterTool: eligible_products + user_conditions 필요

Context 접근이 불필요한 Tool

• PatternAnalyzerTool: 이전 Tool 결과만 사용
• QuestionGeneratorTool: 이전 Tool 결과만 사용
• UserInputTool: 이전 Tool 결과만 사용

기존 파이프라인 방식의 한계: 모든 Tool에 eligible_products와 user_conditions를 파라미터로 전달해야 하는데, 실제로는 2개 Tool만 필요함 → 불필요한 파라미터 전달과 복잡한 의존성 발생

 

파이프라인 구조상 모든 Tool에 데이터를 전달해야 하니 불필요한 파라미터 전달이 강제되고, Tool 간 의존성이 복잡하게 얽히면서 디버깅이 사실상 불가능한 상태가 되었다.

 

더 큰 문제는 멀티 세션 환경이다. 동시에 여러 사용자가 시스템을 사용할 때 각자의 eligible_products와 user_conditions 데이터가 섞이거나 덮어쓰여지는 상황이 발생했다.

 

수많은 구글링 결과 Python의 ContextVar를 도입하여 이 모든 문제를 구조적으로 해결했다.

문제 정의: 기존 접근법의 구조적 한계

시스템 구성

5개 Agent로 구성된 파이프라인

1. EligibilityAgent: 자격 조건 필터링
2. QuestionAgent: 우대조건 역질문 생성 <- 여기서 전역 데이터 필요함
3. StrategyAgent: 전략 시나리오 생성
4. ComparatorAgent: 전략 비교 분석
5. FormatterAgent: 최종 출력 포맷팅

기존 방식의 실패 분석

문제코드: 파라미터 체인 전달

def execute_pipeline(user_input):
    eligible_products = eligibility_agent.execute(user_input)
    questions = question_agent.execute(eligible_products, user_input)
    strategies = strategy_agent.execute(eligible_products, questions, user_input)

결과: Agent 증가 시 파라미터 폭증으로 유지보수성 급격히 저하

해결책: ContextVar 기반 상태 관리

선택 근거

ContextVar는 다음 요구사항을 완전히 충족했다

• Thread-safe 보장
• 비동기 환경 지원
• Context별 완전 격리
• 타입 안전성

핵심 구현

Context 설계의 핵심은 각 데이터 타입별로 독립적인 ContextVar 인스턴스를 생성하는 것이다. 각 ContextVar는 고유한 식별자와 기본값을 가지며, 타입 힌팅을 통해 런타임 안전성을 보장한다.

from contextvars import ContextVar
from typing import Any
from schemas.agent_responses import SimpleProduct
from schemas.eligibility_conditions import EligibilityConditions

class QuestionAgentContext:
    def __init__(self):
        # 각 ContextVar는 독립적인 네임스페이스를 가짐
        self.eligible_products_ctx: ContextVar[list[SimpleProduct]] = ContextVar(
            "eligible_products", default=[]
        )
        self.user_conditions_ctx: ContextVar[EligibilityConditions | None] = ContextVar(
            "user_conditions", default=None
        )
        self.session_id_ctx: ContextVar[str] = ContextVar("session_id", default="")

    def set_eligible_products(self, products: list[SimpleProduct]) -> None:
        """
        EligibilityAgent에서 필터링된 통장 목록을 Context에 저장
        이후 QuestionAgent의 모든 Tool에서 접근 가능해짐
        """
        self.eligible_products_ctx.set(products)

    def get_eligible_products(self) -> list[SimpleProduct]:
        """
        현재 실행 컨텍스트의 통장 목록 조회
        멀티 세션 환경에서도 각 세션의 데이터만 반환됨
        """
        return self.eligible_products_ctx.get()

    def set_user_conditions(self, conditions: EligibilityConditions) -> None:
        """
        사용자 입력 조건을 Context에 저장
        Tool 체인 전반에서 참조 가능
        """
        self.user_conditions_ctx.set(conditions)

    def get_user_conditions(self) -> EligibilityConditions | None:
        """
        저장된 사용자 조건 조회
        None 반환 시 조건이 설정되지 않은 상태
        """
        return self.user_conditions_ctx.get()

Agent 통합

QuestionAgent 초기화 시점에 Context 인스턴스를 생성하고, 이를 모든 Tool에 전달한다. 이렇게 하면 각 Tool이 필요한 시점에 Context에서 데이터를 조회할 수 있다.

class QuestionAgent:
    def __init__(self, llm: BaseLanguageModel, test_mode: bool = True):
        # Agent별 독립적인 Context 생성
        self.agent_ctx = QuestionAgentContext()
        # 모든 Tool에 동일한 Context 인스턴스 전달
        self.tools = QuestionTools.get_tools(llm, test_mode, self.agent_ctx)

Tool 레벨 데이터 접근

각 Tool은 초기화 시점에 Context 참조를 받아 저장하고, 실행 시점에 필요한 데이터를 Context에서 조회한다. 파라미터로 데이터를 전달받을 필요가 없어진다.

class ResponseFormatterTool:
    def __init__(self, agent_context: QuestionAgentContext):
        # Tool 생성 시 Context 참조 저장
        self.context = agent_context
    
    def _format_response(self, user_input_result: UserInputResult) -> QuestionSuccessResponse:
        # 실행 시점에 Context에서 필요한 데이터 조회
        # 다른 Tool의 파라미터와 무관하게 독립적으로 접근
        eligible_products = self.context.get_eligible_products()
        user_conditions = self.context.get_user_conditions()
        
        return QuestionSuccessResponse(
            result_products=eligible_products,
            user_responses=user_input_result.responses,
            user_conditions=user_conditions
        )

기술적 분석: Thread-Safe 매커니즘

ContextVar 격리 원리

ContextVar의 핵심은 각 실행 컨텍스트(execution context)별로 독립적인 값을 유지한다는 점이다. 같은 ContextVar 인스턴스라도 다른 실행 컨텍스트에서는 완전히 다른 값을 가질 수 있다.

# 세션 A에서 실행
context.set_eligible_products([product_a1, product_a2])

# 세션 B에서 동시 실행
context.set_eligible_products([product_b1, product_b2])

# 각 세션에서 get 호출 시:
# 세션 A: [product_a1, product_a2] 반환
# 세션 B: [product_b1, product_b2] 반환
# 데이터 충돌 없음

이는 전역 변수나 클래스 속성과는 완전히 다른 동작이다. 전역 변수를 사용했다면 마지막에 설정된 값으로 모든 세션이 덮어쓰여졌을 것이다.

타입 안전성 구현

ContextVar 생성 시 Generic 타입을 명시하면 IDE와 타입 체커가 완전한 지원을 제공한다. 이는 런타임 에러를 개발 단계에서 사전 차단하는 효과가 있다.

# 타입 명시로 안전성 확보
self.eligible_products_ctx: ContextVar[list[SimpleProduct]] = ContextVar(
    "eligible_products", default=[]
)

# IDE에서 자동완성과 타입 검증 지원
products = self.eligible_products_ctx.get()  # list[SimpleProduct] 타입으로 추론

모니터링 및 디버깅

Context 상태를 실시간으로 추적할 수 있는 디버깅 메서드를 구현했다. 이는 개발 과정에서 데이터 흐름을 파악하고 문제를 진단하는 데 필수적이다.

def get_context_info(self) -> dict[str, Any]:
    """
    현재 Context 상태 정보 조회
    개발/디버깅 시 데이터 흐름 추적용
    """
    return {
        "eligible_products_count": len(self.eligible_products_ctx.get()),
        "has_user_conditions": self.user_conditions_ctx.get() is not None,
        "session_id": self.session_id_ctx.get(),
        "context_status": "active" if self.session_id_ctx.get() else "empty",
    }

실제 운영에서는 이 정보를 로깅하여 Agent 간 데이터 전달 상태를 모니터링한다.

성과 측정

정량적 개선 지표

동시 사용자 10명 환경 테스트 결과:

• 데이터 충돌: 0건 (이전 방식 대비 100% 개선)
• 메모리 사용량: 15% 감소
• 응답 시간: 2.3초 → 2.1초 (9% 개선)
• 코드 복잡도: Tool 간 파라미터 전달 완전 제거

아키텍처 개선

이전: Agent → Tool 파라미터 체인 (N² 복잡도) 현재: Context 중앙집중 관리 (O(1) 접근)

중앙 집중형 상태관리

ContextVar 기반 Multi-Agent 상태 관리는 복잡한 시스템에서 데이터 공유 문제의 근본적 해결책이다. Thread-safe 보장과 동시에 코드 복잡도를 대폭 줄일 수 있었고, 확장성 있는 아키텍처 구축이 가능했다. Multi-Agent 시스템 구축 시 기존 방식의 한계를 인식하고 있다면, ContextVar 도입을 적극 검토하라. 초기 설정 비용 대비 얻는 안정성과 성능 이득이 명확하다.

 

알파코캠퍼스에서 진행한 이 프로젝트에서 많은걸 배울수 있어서 소중한 시간이었다!

---

이 글은 실제 파킹통장 추천 AI 프로젝트를 진행하면서 겪은 시행착오를 바탕으로 작성되었습니다.
전체 코드는 GitHub에서 확인할 수 있습니다.