Sweets Vault — 실물 하드웨어와 연동한 멀티모달 Gemini 에이전트 구축기

load_dotenv()
project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
location = os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
os.environ["GOOGLE_CLOUD_PROJECT"] = project_id
os.environ["GOOGLE_CLOUD_LOCATION"] = location
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"

vertexai.init(project=project_id, location=location)

AGENT_LANGUAGE = os.getenv("AGENT_LANGUAGE", "en")

root_agent = Agent(
    name="root_agent",
    model=Gemini(
        model="gemini-2.5-flash",
        retry_options=types.HttpRetryOptions(attempts=3),
    ),
    instruction=load_prompt_from_file(f"sweet-vault-agent-{AGENT_LANGUAGE}.txt"),
    tools=[get_progress, complete_task, unlock_drawer],
)

def _get_task_status(user_key: str, task_id: str, tool_context: ToolContext) -> bool:
    state_key = f"user_tasks_{user_key}_{task_id}"
    return tool_context.state.get(state_key, False)


def _set_task_status(user_key: str, task_id: str, is_done: bool, tool_context: ToolContext):
    target_key = f"user_tasks_{user_key}_{task_id}"
    tool_context.state[target_key] = is_done

    all_sync_updates = {}
    for name in user_names:
        u_key = name.lower()
        for t_id in TASKS_CONFIG:
            key = f"user_tasks_{u_key}_{t_id}"
            all_sync_updates[key] = tool_context.state.get(key, False)

    tool_context.state.update(all_sync_updates)
    logging.info(f"Synchronized all task state values. Updated {target_key} to {is_done}")

def get_progress(user_name: str, tool_context: ToolContext) -> str:
    user_key = user_name.lower()

    status_msg = f"Progress for {user_name}:\n"
    for task_id, desc in TASKS_CONFIG.items():
        is_done = _get_task_status(user_key, task_id, tool_context)
        state_str = "✅ DONE" if is_done else "❌ PENDING"
        status_msg += f"- [{task_id}] {desc}: {state_str}\n"

    return status_msg

def complete_task(user_name: str, task_id: str, tool_context: ToolContext) -> str:
    user_key = user_name.lower()

    if task_id in TASKS_CONFIG:
        _set_task_status(user_key, task_id, True, tool_context)
    else:
        return f"Error: Task ID '{task_id}' not found."

    all_complete = True
    remaining = []
    for t_id in TASKS_CONFIG:
        if not _get_task_status(user_key, t_id, tool_context):
            all_complete = False
            remaining.append(t_id)

    if all_complete:
        return (
            f"SUCCESS: All tasks completed for {user_name}! "
            "You may now unlock the drawer."
        )

    return (
        f"Task {task_id} marked as DONE. "
        f"Remaining tasks: {', '.join(remaining)}."
    )

user_names = ["Maria", "Jan"] if AGENT_LANGUAGE == "pl" else ["Mary", "James"]
hw_interface = HardwareInterface(user_names)

def unlock_drawer(id: int, user_name: str) -> str:
    if id in [0, 1]:
        hw_interface.unlock_drawer(id)
        return f"Drawer {id} unlocked for {user_name}"

    return "Drawer not found"

You are a friendly, cheerful, and helpful AI assistant, the guardian of the "Sweets Vault."

### MAIN RULES:
1. LANGUAGE: You speak ONLY AND EXCLUSIVELY IN ENGLISH.
2. USERS:
   - Mary (girl, 7 years old) -> Assigned drawer ID: 0
   - James (boy, 7 years old) -> Assigned drawer ID: 1
3. PERSONALITY: You are enthusiastic, warm, and supportive.

### TASK VERIFICATION PROCESS:
1. STATE IDENTIFICATION: ALWAYS use get_progress(user_name) first.
2. REPORTING: The child reports completing a task (A or B).
3. VERIFICATION: Conduct a rigorous verification (camera/questions).
4. CREDITING: If verification is successful, use complete_task(user_name, task_id).
   ONLY IF the response is "SUCCESS: All tasks completed...", use unlock_drawer.

Task A: Reading — Ask child to show the page to camera, then ask a comprehension question.
Task B: Calligraphy — Ask child to show the completed workbook page to camera.