### 질문 > 내가 개발 중인 에이전트는 다중 작업 workflow를 가지고 있다. 특정 stage에서도 여러 작업들이 순차적 혹은 병렬적으로 실행된다. > > 다음을 가정하라: > 특정 작업 stage에서 Task A와 Task B가 순차적으로 실행된다. Task A와 Task B에서 모두 공통된 LLM(gemini-3.1-flash-lite-preview)을 사용한다. > > 위의 가정 하에서 Task_B는 입력개체로서 Task A 입력의 일부 파일, 그리고 Task A의 결과물 파일을 입력값으로 사용한다. LLM 비용/속도 측면에서 Task A와 Task B 사이에서 google이 제공하는 context caching 작업을 끼워넣으려고 한다. 어떻게 작업을 하면 되는가? ### 답변 Task A와 Task B가 공통된 입력 파일을 공유하는 워크플로우라면, Gemini의 Context Caching을 도입하여 비용을 절감하고 응답 속도(TTFT, Time To First Token)를 높이기에 아주 이상적인 시나리오입니다. Context Caching을 설계할 때 가장 중요하게 기억해야 할 점은 **"생성된 캐시는 불변(Immutable) 객체"**라는 것입니다. 즉, 한 번 만든 캐시에 나중에 Task A의 결과물을 '추가'할 수는 없습니다. 따라서 다음과 같은 워크플로우로 아키텍처를 설계하셔야 합니다. ------ ## 💡 Context Caching 적용 워크플로우 **1. 공통 입력 데이터 식별 및 캐시 생성 (Before Task A)** Task A와 Task B가 공통으로 사용하는 '일부 파일'들을 모아 하나의 캐시(Cache)로 생성합니다. 이때 캐시의 수명(TTL, Time To Live)을 해당 Stage가 실행되는 여유 시간(예: 10분~15분) 정도로 설정합니다. **2. Task A 실행** - **입력:** `[1번에서 만든 캐시 객체]` + `[Task A에만 필요한 추가 파일/프롬프트]` - **출력:** Task A의 결과물 파일 (Output A) **3. Task B 실행** - **입력:** `[1번에서 만든 동일한 캐시 객체]` + `[Task A의 결과물 (Output A)]` + `[Task B 전용 프롬프트]` - **출력:** Task B의 최종 결과물 **4. 캐시 삭제 (After Task B)** 작업 Stage가 끝나면 생성해 둔 캐시를 명시적으로 삭제합니다. (TTL이 지나면 자동 삭제되지만, 스토리지 유지 비용을 최소화하기 위해 즉시 삭제하는 것이 좋습니다.) ------ ## 💻 Python 구현 논리 (의사 코드 및 핵심 API) Google GenAI SDK를 사용할 경우 대략적인 작업 흐름은 다음과 같습니다. Python ``` import google.generativeai as genai from google.generativeai import caching import datetime # 1. 공통 파일 업로드 common_file = genai.upload_file("path/to/shared_input.pdf") # 2. 공통 컨텍스트 캐시 생성 (Task A와 B가 공유할 데이터) cache = caching.CachedContent.create( model='models/gemini-1.5-flash-002', # 혹은 사용하시는 모델 display_name='shared_stage_cache', system_instruction='당신은 다중 작업 워크플로우의 AI 에이전트입니다.', contents=[common_file], ttl=datetime.timedelta(minutes=15), # 15분 후 자동 삭제 ) # 3. Task A 실행 model_A = genai.GenerativeModel.from_cached_content(cached_content=cache) response_A = model_A.generate_content("Task A를 수행해줘. 이 파일에서 주요 키워드를 추출해.") task_a_output = response_A.text # 4. Task B 실행 (동일한 캐시 재사용 + Task A 결과물 주입) model_B = genai.GenerativeModel.from_cached_content(cached_content=cache) response_B = model_B.generate_content([ f"이전 작업의 결과물은 다음과 같아: {task_a_output}", "이 결과물과 캐시된 문서를 바탕으로 Task B(최종 요약 보고서 작성)를 수행해줘." ]) task_b_output = response_B.text # 5. 작업 완료 후 캐시 즉시 삭제 (비용 절감) cache.delete() ``` *(참고: `gemini-3.1-flash-lite-preview`와 같은 최신/프리뷰 모델의 경우, API 버전이나 SDK 업데이트 상태에 따라 캐싱 지원 여부 및 모델명이 다를 수 있으니 사용 전 공식 문서를 확인해 주세요.)* ------ ## ⚠️ 실무 적용 시 주의사항 (Gotchas) - **최소 토큰 수 제한:** Gemini의 Context Caching은 보통 **32,768 토큰 이상**의 컨텍스트에 대해서만 작동합니다. 공유하는 입력 파일의 크기가 이보다 작다면 캐시 생성이 거부되거나, 오히려 일반 호출이 더 저렴하고 빠를 수 있습니다. - **스토리지 비용:** 캐시된 토큰은 호출할 때마다 발생하는 입력(Input) 비용은 크게 줄여주지만, 캐시가 유지되는 동안 분/시간 단위로 '스토리지 보관 비용'이 발생합니다. Task B가 끝나는 즉시 캐시를 폐기(`delete`)하는 로직을 반드시 `try-finally` 구문 등으로 안전하게 구현해야 합니다. - **Task A 출력물의 크기:** Task A의 결과물은 캐시되지 않고 매번 Task B 호출 시 동적 입력 토큰으로 계산됩니다. 만약 Task A의 결과물이 너무 방대하고 이후 Task C, Task D에서도 계속 쓰인다면, Task B 단계에서 기존 캐시를 삭제하고 Task A 결과물까지 포함한 **새로운 캐시를 굽는(Baking) 것**이 나을 수도 있습니다. 현재 개발 중인 에이전트 환경(Python, Node.js 등)에 맞춰 더 구체적인 코드 예시나 에러 핸들링 로직을 작성해 드릴까요? --- ### 질문 > 에러 핸들링 로직을 포함한 Python 코드 예시를 작성해줘. ### 답변 실무 환경에서 에러 핸들링은 캐시 누수로 인한 불필요한 비용 발생을 막고, 예상치 못한 API 오류(예: 토큰 수 미달, 할당량 초과)에 유연하게 대처하기 위해 필수적입니다. 가장 중요한 부분은 작업 성공 여부와 관계없이 **반드시 캐시와 업로드된 파일을 삭제하도록 `finally` 블록을 구성**하는 것입니다. 아래는 Google GenAI SDK를 활용하여 에러 핸들링 로직을 적용한 완전한 형태의 Python 예시 코드입니다. ## 💻 Context Caching 통합 및 에러 핸들링 구현 Python ``` import os import datetime import google.generativeai as genai from google.generativeai import caching from google.api_core import exceptions # API 키 설정 (환경 변수 사용 권장) genai.configure(api_key=os.environ.get("GEMINI_API_KEY")) def run_multi_task_workflow(shared_file_path: str): # 사용자가 지정한 모델 model_name = "models/gemini-3.1-flash-lite-preview" uploaded_file = None cache = None try: # 1. 공통 파일 업로드 print("1. 공통 파일 업로드 중...") uploaded_file = genai.upload_file(shared_file_path) # 2. Context Cache 생성 print("2. Context Cache 생성 중...") cache = caching.CachedContent.create( model=model_name, display_name="workflow_shared_cache", system_instruction="당신은 다중 작업 워크플로우를 처리하는 AI 에이전트입니다.", contents=[uploaded_file], ttl=datetime.timedelta(minutes=15) # 15분 후 자동 삭제되도록 안전장치 설정 ) print(f" -> 캐시 생성 완료 (이름: {cache.name})") # 3. Task A 실행 print("\n3. Task A 실행 중...") model_a = genai.GenerativeModel.from_cached_content(cached_content=cache) response_a = model_a.generate_content("Task A: 이 문서의 핵심 요약 3가지를 추출해 주세요.") task_a_output = response_a.text print(f" -> [Task A 결과]:\n{task_a_output}") # 4. Task B 실행 (캐시 재사용 + Task A 결과물 주입) print("\n4. Task B 실행 중...") model_b = genai.GenerativeModel.from_cached_content(cached_content=cache) prompt_b = [ f"이전 작업(Task A)의 결과물은 다음과 같습니다:\n{task_a_output}\n\n", "Task B: 캐시된 원본 문서와 위의 Task A 결과물을 바탕으로, 최종 분석 보고서를 작성해 주세요." ] response_b = model_b.generate_content(prompt_b) task_b_output = response_b.text print(f" -> [Task B 결과]:\n{task_b_output}") return task_b_output except exceptions.InvalidArgument as e: # 캐싱 최소 토큰 수(일반적으로 32k)를 채우지 못했을 때 자주 발생합니다. print(f"\n[오류] 잘못된 요청 (토큰 수 미달 등): {e}") # 필요하다면 여기서 캐싱을 사용하지 않는 일반 API 호출 로직으로 Fallback 할 수 있습니다. except exceptions.ResourceExhausted as e: print(f"\n[오류] API 할당량 초과: {e}") except exceptions.GoogleAPIError as e: print(f"\n[오류] Google API 통신 오류 발생: {e}") except Exception as e: print(f"\n[오류] 예기치 않은 시스템 오류 발생: {e}") finally: # 5. 리소스 정리 (성공/실패 여부와 무관하게 실행) print("\n5. 리소스 정리 중...") # 캐시 삭제 (스토리지 비용 발생 방지) if cache: try: cache.delete() print(f" -> 캐시 삭제 완료: {cache.name}") except Exception as e: print(f" -> [경고] 캐시 삭제 실패 (수동 확인 필요): {e}") # 파일 삭제 (File API 스토리지 정리) if uploaded_file: try: genai.delete_file(uploaded_file.name) print(f" -> 업로드된 원본 파일 삭제 완료: {uploaded_file.name}") except Exception as e: print(f" -> [경고] 파일 삭제 실패: {e}") # 실행 예시 # workflow_result = run_multi_task_workflow("sample_document.pdf") ``` ------ ## 💡 코드 핵심 포인트 - **`google.api_core.exceptions` 활용:** `InvalidArgument` 예외 처리를 통해 캐시 생성에 필요한 **최소 토큰 조건(보통 32,768 토큰)**을 만족하지 못했을 때 프로그램이 완전히 죽는 것을 방지합니다. - **Fallback 전략 추가 가능성:** `InvalidArgument` 예외가 발생한 경우, 파일 크기가 작아서 발생한 문제일 확률이 높으므로 이 예외 블록 안에서 캐시를 사용하지 않는 `genai.GenerativeModel()`을 직접 호출하도록 Fallback 로직을 짜는 것도 좋은 설계입니다. - **이중 `try-except` (Finally 블록 내부):** 리소스 정리 중 하나(예: 캐시 삭제)가 실패하더라도 다른 하나(예: 파일 삭제)가 정상적으로 실행될 수 있도록 방어적으로 코딩했습니다. 에이전트가 처리하는 입력 파일의 크기가 작아서(32,000 토큰 미만) 캐시 생성에 실패할 경우를 대비해, **일반 API 호출로 자동 전환(Fallback)되는 로직**도 추가로 작성해 드릴까요? --- ### 질문 > 에이전트가 처리하는 입력 파일의 크기가 작아서(32,000 토큰 미만) 캐시 생성에 실패할 경우를 대비해, 일반 API 호출로 자동 전환(Fallback)되는 로직도 추가로 작성하라. ### 답변 입력 문서의 토큰 수가 적어 캐시 생성이 거부되는 것은 실무 에이전트를 개발할 때 굉장히 자주 마주치는 상황입니다. 이를 예외 처리로 끝내지 않고 **일반 API 호출로 자연스럽게 전환(Fallback)** 되도록 설계하면, 파일 크기에 구애받지 않는 훨씬 안정적이고 유연한 워크플로우를 만들 수 있습니다. 이 Fallback 로직의 핵심은 **캐시 생성에 실패했을 때 분기 플래그(예: `use_cache = False`)를 설정하고, 이후 Task A와 Task B에서 입력 프롬프트 배열에 원본 파일 객체를 직접 끼워 넣는 것**입니다. 다음은 Fallback 메커니즘이 완벽하게 통합된 개선된 파이썬 코드입니다. ------ ## 💻 Context Caching + 자동 Fallback 통합 코드 Python ``` import os import datetime import google.generativeai as genai from google.generativeai import caching from google.api_core import exceptions # API 키 설정 (환경 변수 사용 권장) genai.configure(api_key=os.environ.get("GEMINI_API_KEY")) def run_multi_task_workflow_with_fallback(shared_file_path: str): model_name = "models/gemini-3.1-flash-lite-preview" system_instruction = "당신은 다중 작업 워크플로우를 처리하는 AI 에이전트입니다." uploaded_file = None cache = None use_cache = False # Fallback 분기를 위한 플래그 try: # 1. 공통 파일 업로드 (캐싱 및 일반 호출 모두에 사용됨) print("1. 공통 파일 업로드 중...") uploaded_file = genai.upload_file(shared_file_path) # 2. Context Cache 생성 시도 print("2. Context Cache 생성 시도 중...") try: cache = caching.CachedContent.create( model=model_name, display_name="workflow_shared_cache", system_instruction=system_instruction, contents=[uploaded_file], ttl=datetime.timedelta(minutes=15) ) print(f" -> 캐시 생성 완료 (이름: {cache.name})") use_cache = True # 캐시 생성 성공 시 플래그 활성화 except exceptions.InvalidArgument as e: # 32k 토큰 미만 등의 이유로 캐시 생성이 거부된 경우 Fallback 실행 print(f" -> [알림] 파일 크기 미달 등으로 캐시 생성 생략. 일반 API 호출로 전환(Fallback)합니다. ({e})") use_cache = False # 3. Task A 실행 print("\n3. Task A 실행 중...") task_a_prompt = "Task A: 이 문서의 핵심 요약 3가지를 추출해 주세요." if use_cache: # 캐시가 있을 경우: 파일은 이미 캐시에 포함되어 있으므로 텍스트 프롬프트만 전달 model_a = genai.GenerativeModel.from_cached_content(cached_content=cache) response_a = model_a.generate_content(task_a_prompt) else: # Fallback (캐시 없음): 시스템 프롬프트를 포함하여 모델을 초기화하고, 파일 객체를 함께 전달 model_a = genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction) response_a = model_a.generate_content([uploaded_file, task_a_prompt]) task_a_output = response_a.text print(f" -> [Task A 결과]:\n{task_a_output}") # 4. Task B 실행 print("\n4. Task B 실행 중...") task_b_instruction = "Task B: 원본 문서와 위의 Task A 결과물을 바탕으로, 최종 분석 보고서를 작성해 주세요." if use_cache: # 캐시가 있을 경우: 원본 문서는 캐시에 있으므로 Task A 결과와 Task B 지시문만 전달 model_b = genai.GenerativeModel.from_cached_content(cached_content=cache) prompt_b = [ f"이전 작업(Task A)의 결과물은 다음과 같습니다:\n{task_a_output}\n\n", task_b_instruction ] response_b = model_b.generate_content(prompt_b) else: # Fallback (캐시 없음): 원본 파일 객체 + Task A 결과 + Task B 지시문을 모두 전달 model_b = genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction) prompt_b = [ uploaded_file, f"\n\n이전 작업(Task A)의 결과물은 다음과 같습니다:\n{task_a_output}\n\n", task_b_instruction ] response_b = model_b.generate_content(prompt_b) task_b_output = response_b.text print(f" -> [Task B 결과]:\n{task_b_output}") return task_b_output except Exception as e: print(f"\n[오류] 워크플로우 실행 중 예기치 않은 오류 발생: {e}") return None finally: # 5. 리소스 정리 (성공/실패/Fallback 여부와 무관하게 무조건 실행) print("\n5. 리소스 정리 중...") if cache: try: cache.delete() print(f" -> 캐시 삭제 완료: {cache.name}") except Exception as e: print(f" -> [경고] 캐시 삭제 실패: {e}") if uploaded_file: try: genai.delete_file(uploaded_file.name) print(f" -> 업로드된 원본 파일 삭제 완료: {uploaded_file.name}") except Exception as e: print(f" -> [경고] 파일 삭제 실패: {e}") # 실행 예시 # result = run_multi_task_workflow_with_fallback("sample_document.pdf") ``` ------ ## 💡 로직 설계 포인트 - **중복 업로드 방지:** 파일 크기가 작더라도 File API(`genai.upload_file`)를 통해 업로드된 파일의 URI 객체는 일반 호출의 `generate_content` 배열 안에 그대로 사용할 수 있습니다. 따라서 캐싱이 실패하더라도 파일을 다시 읽거나 재업로드할 필요가 없습니다. - **시스템 프롬프트 유지:** Fallback으로 전환되어 일반 `GenerativeModel`을 생성할 때, 캐시 생성 시 사용하려 했던 `system_instruction`을 잊지 않고 주입해주어야 에이전트의 페르소나가 유지됩니다. - **유연한 페이로드(Payload) 구성:** `generate_content`는 리스트 형태로 여러 모달리티(파일 객체, 텍스트 등)를 순차적으로 받을 수 있습니다. Fallback 시에는 `[uploaded_file, 텍스트_프롬프트]` 형태로 구성하여 모델이 문맥을 온전히 이해하도록 구성했습니다. 코드를 에이전트 파이프라인에 이식하실 때, 각 Task의 출력(Output)을 단순히 문자열로 주고받는 것 외에 JSON(Structured Output) 형태로 파싱하는 로직이 추가로 필요하신가요? --- ### 질문 > 각 Task의 출력(Output)을 단순히 문자열로 주고받는 것 외에 JSON(Structured Output) 형태로 파싱하는 로직이 추가로 필요하다. 이를 반영하여 위의 python code 예제를 새로 작성하라. ### 답변 에이전트 워크플로우에서 각 Task의 결과를 단순 문자열이 아닌 **JSON(Structured Output) 형태**로 주고받는 것은 선택이 아닌 필수입니다. 결과를 파싱할 때 발생할 수 있는 환각(Hallucination)이나 포맷 오류를 방지하고, 다음 Task로 데이터를 안정적으로 넘기기 위해 아주 훌륭한 접근입니다. Gemini API에서는 `response_mime_type="application/json"` 설정과 함께 **Pydantic 모델(스키마)**을 `response_schema`로 전달하여 출력 형태를 강제할 수 있습니다. 이전의 캐싱 및 Fallback 로직에 **Pydantic을 활용한 JSON 구조화 로직**을 추가하여 코드를 전면 개편했습니다. ------ ## 💻 Context Caching + Fallback + JSON Structured Output 통합 코드 Python ``` import os import json import datetime import google.generativeai as genai from google.generativeai import caching from google.api_core import exceptions from pydantic import BaseModel, Field # API 키 설정 (환경 변수 사용 권장) genai.configure(api_key=os.environ.get("GEMINI_API_KEY")) # ========================================== # 📊 1. JSON 출력을 위한 Pydantic 스키마 정의 # ========================================== class TaskAOutput(BaseModel): key_summaries: list[str] = Field( description="문서에서 추출한 핵심 요약 3가지", min_items=1, max_items=3 ) extracted_keywords: list[str] = Field( description="문서의 주요 키워드 목록" ) class TaskBOutput(BaseModel): report_title: str = Field(description="최종 분석 보고서의 제목") executive_summary: str = Field(description="Task A의 결과를 바탕으로 한 전체 요약") detailed_analysis: str = Field(description="원본 문서와 Task A 결과를 종합한 심층 분석 내용") conclusion: str = Field(description="최종 결론 및 인사이트") # ========================================== # ⚙️ 2. 메인 워크플로우 함수 # ========================================== def run_structured_multi_task_workflow(shared_file_path: str): model_name = "models/gemini-3.1-flash-lite-preview" system_instruction = "당신은 다중 작업 워크플로우를 처리하고 반드시 지정된 JSON 포맷으로만 응답하는 AI 에이전트입니다." uploaded_file = None cache = None use_cache = False try: # [공통] 파일 업로드 print("1. 공통 파일 업로드 중...") uploaded_file = genai.upload_file(shared_file_path) # [캐시] Context Cache 생성 시도 print("2. Context Cache 생성 시도 중...") try: cache = caching.CachedContent.create( model=model_name, display_name="workflow_json_cache", system_instruction=system_instruction, contents=[uploaded_file], ttl=datetime.timedelta(minutes=15) ) print(f" -> 캐시 생성 완료 (이름: {cache.name})") use_cache = True except exceptions.InvalidArgument as e: print(f" -> [알림] 파일 크기 미달로 일반 API 호출로 전환(Fallback)합니다.") use_cache = False # ========================================== # 🚀 3. Task A 실행 (JSON 출력 강제) # ========================================== print("\n3. Task A 실행 중 (JSON 파싱)...") task_a_prompt = "Task A: 이 문서의 핵심 요약 3가지와 주요 키워드를 추출해 주세요." # Task A를 위한 JSON Generation Config 설정 config_a = genai.GenerationConfig( response_mime_type="application/json", response_schema=TaskAOutput ) model_a = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction)) prompt_payload_a = [task_a_prompt] if use_cache else [uploaded_file, task_a_prompt] response_a = model_a.generate_content(prompt_payload_a, generation_config=config_a) # 응답 텍스트는 완벽한 JSON 문자열입니다. 파이썬 딕셔너리로 변환합니다. task_a_dict = json.loads(response_a.text) print(f" -> [Task A JSON 결과]:\n{json.dumps(task_a_dict, indent=2, ensure_ascii=False)}") # ========================================== # 🚀 4. Task B 실행 (Task A JSON 주입 -> JSON 출력) # ========================================== print("\n4. Task B 실행 중 (JSON 파싱)...") # Task A의 파싱된 딕셔너리를 다시 문자열 형태로 깔끔하게 프롬프트에 주입 task_b_instruction = f""" Task B: 원본 문서와 이전 작업(Task A)의 구조화된 결과물을 바탕으로 최종 분석 보고서를 작성해 주세요. [Task A 결과 데이터]: {json.dumps(task_a_dict, ensure_ascii=False)} """ # Task B를 위한 JSON Generation Config 설정 config_b = genai.GenerationConfig( response_mime_type="application/json", response_schema=TaskBOutput ) model_b = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction)) prompt_payload_b = [task_b_instruction] if use_cache else [uploaded_file, task_b_instruction] response_b = model_b.generate_content(prompt_payload_b, generation_config=config_b) task_b_dict = json.loads(response_b.text) print(f" -> [Task B JSON 결과]:\n{json.dumps(task_b_dict, indent=2, ensure_ascii=False)}") return task_b_dict except json.JSONDecodeError as e: print(f"\n[오류] JSON 파싱 실패 (모델이 JSON 형식을 어겼습니다): {e}") return None except Exception as e: print(f"\n[오류] 워크플로우 실행 중 예기치 않은 오류 발생: {e}") return None finally: # 5. 리소스 정리 print("\n5. 리소스 정리 중...") if cache: try: cache.delete(); print(f" -> 캐시 삭제 완료") except: pass if uploaded_file: try: genai.delete_file(uploaded_file.name); print(f" -> 원본 파일 삭제 완료") except: pass # 실행 예시 # final_json_report = run_structured_multi_task_workflow("sample_document.pdf") ``` ------ ## 💡 구조화된 출력(Structured Output) 적용 포인트 1. **`response_schema`의 강력함:** Pydantic 클래스(`TaskAOutput`, `TaskBOutput`)를 `genai.GenerationConfig`에 넘겨주면, Gemini 모델이 백엔드 단에서 문법적으로 완벽한 JSON을 생성하도록 강제됩니다. 문자열 파싱 로직이나 정규표현식을 쓸 필요가 없어집니다. 2. **에이전트 간 명확한 데이터 핸드오프(Handoff):** Task A의 출력인 `response_a.text`는 순수한 JSON 문자열입니다. 이를 `json.loads()`로 파이썬 딕셔너리로 바꾼 뒤, Task B의 프롬프트에 `json.dumps()`를 통해 명확한 키-값(Key-Value) 형태로 주입했습니다. 이렇게 하면 Task B의 LLM이 데이터를 훨씬 정확하게 인식합니다. 3. **`Field(description="...")`의 활용:** Pydantic의 `description` 속성은 단순한 주석이 아닙니다. 이 설명 자체가 내부적으로 **프롬프트의 일부로 모델에게 전달**되어, 각 필드에 어떤 내용을 채워야 하는지 모델을 직접적으로 가이드합니다. 이 코드를 실행하다 보면 가끔 모델이 스키마를 미세하게 어기거나(극히 드묾), 네트워크 지연이 발생할 수 있습니다. **안정성을 더 높이기 위해 재시도(Retry) 로직이나 Timeout 설정을 추가로 구성해 드릴까요?** --- ### 질문 > 안정성을 더 높이기 위해 재시도(Retry) 로직이나 Timeout 설정을 추가로 구성하여 python 예제 코드를 재작성하라. ### 답변 실무 환경에서 LLM API를 호출할 때는 네트워크 지연(Timeout), 일시적인 서버 오류(500, 503), 혹은 API 호출 제한(Rate Limit) 등의 예기치 않은 상황이 빈번하게 발생합니다. 이를 방어하기 위해 **재시도(Retry) 알고리즘(Exponential Backoff 적용)**과 **Timeout(요청 시간 초과) 설정**을 통합한 최종 완성형 코드를 작성했습니다. 외부 라이브러리(예: `tenacity`) 없이도 바로 복사해서 사용할 수 있도록 파이썬 기본 모듈만 사용하여 구성했습니다. ------ ## 💻 Context Caching + Fallback + JSON + Retry/Timeout 통합 코드 Python ``` import os import json import time import datetime import google.generativeai as genai from google.generativeai import caching from google.api_core import exceptions as google_exceptions from pydantic import BaseModel, Field # API 키 설정 genai.configure(api_key=os.environ.get("GEMINI_API_KEY")) # ========================================== # 📊 1. JSON 출력을 위한 Pydantic 스키마 # ========================================== class TaskAOutput(BaseModel): key_summaries: list[str] = Field(description="문서에서 추출한 핵심 요약 3가지", min_items=1, max_items=3) extracted_keywords: list[str] = Field(description="문서의 주요 키워드 목록") class TaskBOutput(BaseModel): report_title: str = Field(description="최종 분석 보고서의 제목") detailed_analysis: str = Field(description="원본 문서와 Task A 결과를 종합한 심층 분석 내용") # ========================================== # 🛡️ 2. 안정성을 위한 Retry & Timeout 래퍼 함수 # ========================================== def generate_with_retry(model, prompt_payload, generation_config, max_retries=3, timeout_sec=60): """ Timeout을 적용하고, 일시적인 네트워크/서버 오류 시 지수 백오프(Exponential Backoff)로 재시도합니다. """ for attempt in range(max_retries): try: # request_options를 통해 타임아웃(초)을 설정합니다. response = model.generate_content( prompt_payload, generation_config=generation_config, request_options={"timeout": timeout_sec} ) return response except (google_exceptions.ServiceUnavailable, google_exceptions.DeadlineExceeded, google_exceptions.InternalServerError, google_exceptions.TooManyRequests) as e: print(f" -> [경고] API 통신 지연 또는 일시적 오류 (시도 {attempt + 1}/{max_retries}): {e}") if attempt < max_retries - 1: sleep_time = 2 ** attempt # 1초, 2초, 4초 대기 (Exponential Backoff) print(f" -> {sleep_time}초 후 재시도합니다...") time.sleep(sleep_time) else: print(" -> [실패] 최대 재시도 횟수를 초과했습니다.") raise e # 최종 실패 시 예외를 던져 워크플로우 중단 # ========================================== # ⚙️ 3. 메인 워크플로우 함수 # ========================================== def run_robust_workflow(shared_file_path: str): model_name = "models/gemini-3.1-flash-lite-preview" system_instruction = "당신은 다중 작업 워크플로우를 처리하고 반드시 지정된 JSON 포맷으로 응답하는 AI입니다." uploaded_file = None cache = None use_cache = False try: # [공통] 파일 업로드 print("1. 파일 업로드 중...") uploaded_file = genai.upload_file(shared_file_path) # [캐시] Context Cache 생성 (15분 유지) print("2. 캐시 생성 시도 중...") try: cache = caching.CachedContent.create( model=model_name, display_name="robust_workflow_cache", system_instruction=system_instruction, contents=[uploaded_file], ttl=datetime.timedelta(minutes=15) ) use_cache = True print(f" -> 캐시 성공: {cache.name}") except google_exceptions.InvalidArgument: use_cache = False print(" -> [Fallback] 토큰 수 미달로 일반 호출로 전환합니다.") # ========================================== # 🚀 4. Task A 실행 (Retry + Timeout + JSON) # ========================================== print("\n3. Task A 실행 중...") task_a_prompt = "Task A: 이 문서의 핵심 요약 3가지와 주요 키워드를 추출해 주세요." config_a = genai.GenerationConfig(response_mime_type="application/json", response_schema=TaskAOutput) model_a = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction)) payload_a = [task_a_prompt] if use_cache else [uploaded_file, task_a_prompt] # 래퍼 함수를 통해 안전하게 호출 (Timeout 60초, 재시도 3회) response_a = generate_with_retry(model_a, payload_a, config_a, max_retries=3, timeout_sec=60) task_a_dict = json.loads(response_a.text) print(" -> [Task A 완료]") # ========================================== # 🚀 5. Task B 실행 (Retry + Timeout + JSON) # ========================================== print("\n4. Task B 실행 중...") task_b_prompt = f""" Task B: 원본 문서와 이전 작업(Task A)의 결과물을 바탕으로 최종 분석 보고서를 작성해 주세요. [Task A 결과]: {json.dumps(task_a_dict, ensure_ascii=False)} """ config_b = genai.GenerationConfig(response_mime_type="application/json", response_schema=TaskBOutput) model_b = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction)) payload_b = [task_b_prompt] if use_cache else [uploaded_file, task_b_prompt] # 래퍼 함수를 통해 안전하게 호출 (Timeout 90초 설정 - Task B가 더 오래 걸릴 것을 가정) response_b = generate_with_retry(model_b, payload_b, config_b, max_retries=3, timeout_sec=90) task_b_dict = json.loads(response_b.text) print(" -> [Task B 완료]") return task_b_dict except Exception as e: print(f"\n[치명적 오류] 워크플로우 중단: {e}") return None finally: # [정리] 리소스 해제 print("\n5. 리소스 정리 중...") if cache: try: cache.delete(); print(" -> 캐시 삭제 완료") except: pass if uploaded_file: try: genai.delete_file(uploaded_file.name); print(" -> 파일 삭제 완료") except: pass # 실행 예시 # result = run_robust_workflow("sample_document.pdf") ``` ## 💡 주요 개선 사항 요약 1. **`request_options={"timeout": timeout_sec}` 적용:** - Gemini API 호출 시 응답이 무한정 길어지거나 연결이 끊겨 에이전트가 멈추는(Hang) 현상을 방지합니다. - 복잡한 추론이 들어가는 Task B는 시간을 더 넉넉하게(90초) 주는 등 유연한 설정이 가능해졌습니다. 2. **`generate_with_retry` 헬퍼 함수 구현:** - `DeadlineExceeded`(타임아웃), `ServiceUnavailable`(서버 점검/일시 오류), `TooManyRequests`(초당 호출 제한 초과) 등 복구 가능한 오류만 필터링하여 재시도합니다. - **지수 백오프(Exponential Backoff)**: 첫 실패 시 1초 대기, 두 번째 2초 대기, 세 번째 4초 대기하는 방식으로 서버 부하를 줄이면서 성공 확률을 높입니다. 3. **오류 전파 최적화:** - 최대 재시도 횟수를 넘기면 강제로 예외(`raise e`)를 발생시켜, 비정상적인 상태로 워크플로우가 계속 진행되어 불필요한 비용이 발생하는 것을 차단합니다.