692 lines
34 KiB
Markdown
692 lines
34 KiB
Markdown
### 질문
|
|
|
|
> 내가 개발 중인 에이전트는 다중 작업 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`)를 발생시켜, 비정상적인 상태로 워크플로우가 계속 진행되어 불필요한 비용이 발생하는 것을 차단합니다. |