XSkill: Multimodal Agent의 Experience와 Skill 기반 Continual Learning

# Experience 추출 후 검색 예시 (핵심 로직 스케치)
import json
from openai import OpenAI
import numpy as np

client = OpenAI()

def embed(text: str) -> list[float]:
    res = client.embeddings.create(model="text-embedding-3-small", input=text)
    return res.data[0].embedding

def cosine_sim(a, b):
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

# Experience Bank (JSON)
experience_bank = [
    {"id": "E1", "condition": "When image is upside-down",
     "action": "Rotate image before analysis using img.rotate(180)",
     "embedding": embed("When image is upside-down rotate image before analysis")},
    {"id": "E2", "condition": "When object is too small to identify",
     "action": "Crop and zoom in with code interpreter before searching",
     "embedding": embed("When object is too small crop and zoom in")},
]

def retrieve_experiences(query: str, top_k: int = 3, threshold: float = 0.0):
    q_emb = embed(query)
    scored = [(e, cosine_sim(q_emb, e["embedding"])) for e in experience_bank]
    scored = [(e, s) for e, s in scored if s > threshold]
    scored.sort(key=lambda x: -x[1])
    return [e for e, _ in scored[:top_k]]

def decompose_task(task_description: str, image_context: str) -> list[str]:
    """태스크를 2-3개 subtask 쿼리로 분해"""
    prompt = f"""Decompose this visual task into 2-3 abstract subtask queries for experience retrieval.
Task: {task_description}
Image context: {image_context}
Output JSON array of query strings only."""
    res = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )
    return json.loads(res.choices[0].message.content)

# 사용 예시
task = "What is the prototype of the two mascots in the corner of the picture?"
image_ctx = "Image appears to be rotated 180 degrees, small mascots in corner"

subtask_queries = decompose_task(task, image_ctx)
all_experiences = []
for q in subtask_queries:
    exps = retrieve_experiences(q, top_k=3)
    all_experiences.extend(exps)

# 중복 제거 후 system prompt에 주입
unique_exps = {e['id']: e for e in all_experiences}.values()
injection = "\n".join([f"[{e['id']}] {e['condition']}: {e['action']}" for e in unique_exps])
print("Injected experiences:\n", injection)