Firestore로 AI Agent Instruction과 Semantic View를 계층적으로 관리하는 설계

# instructions/{scope}/workspaces/{workspace_id}
# semantic_views/{scope}/workspaces/{workspace_id}
{
    "name": "예금 분석",
    "description": "예금 관련 시맨틱뷰 모음",
    "owner_id": "hong@sh-bank.com",
    "dept_id": "IT팀",
    "scope": "user",               # "global" | "dept" | "user"
    "tags": ["예금", "분석"],
    "item_count": 3,               # 비정규화: 목록 화면에서 건수 표시용
    "created_at": "2026-03-24T09:00:00Z",
    "updated_at": "2026-03-24T14:30:00Z"
}

# instructions/{scope}/workspaces/{workspace_id}/items/{item_id}
{
    "name": "금융상담 기본 지시",
    "description": "일반 고객 대상 금융 상담 에이전트용 기본 Instruction",
    "content": "당신은 SH은행의 금융 상담 AI 어시스턴트입니다...",
    "content_type": "text",        # "text" | "yaml" | "json"
    "owner_id": "hong@sh-bank.com",
    "workspace_id": "표준_상담",   # 역참조 (collection_group 쿼리용)
    "scope": "global",
    "visibility": "published",     # "draft" | "shared" | "published"
    "promoted_from": "",           # 승격 전 원본 경로
    "assigned_agents": [           # 이 Instruction을 사용 중인 에이전트 목록
        "agent:financial-advisor",
        "agent:loan-consultant"
    ],
    "tags": ["상담", "금융", "기본"],
    "version": 3,
    "created_at": "2026-03-24T09:00:00Z",
    "updated_at": "2026-03-24T14:30:00Z"
}

# semantic_views/{scope}/workspaces/{workspace_id}/views/{view_id}
{
    "name": "SV_예금잔액",
    "description": "일별 전체 예금 잔액 현황",
    "yaml_string": "base_table: banking.daily_deposit_balance\n...",
    "owner_id": "hong@sh-bank.com",
    "workspace_id": "예금분석",
    "scope": "user",
    "visibility": "draft",
    "promoted_from": "",
    "tags": ["예금", "KPI"],
    "version": 1,
    "created_at": "2026-03-24T09:00:00Z",
    "updated_at": "2026-03-24T14:30:00Z"
}

# agent_assignments/{agent_id}/config
{
    "agent_id": "financial-advisor",
    "agent_name": "금융 상담 에이전트",
    "instruction_ref": "instructions/_global/workspaces/표준_상담/items/금융상담_기본",
    "semantic_view_refs": [
        "semantic_views/_global/workspaces/표준_KPI/views/SV_예금잔액",
        "semantic_views/dept:IT팀/workspaces/IT_대시보드/views/SV_서버현황"
    ],
    "assigned_by": "hong@sh-bank.com",
    "override_instruction_ref": "",  # 사용자별 오버라이드 (선택)
    "is_active": true,
    "updated_at": "2026-03-24T14:30:00Z"
}

from google.cloud import firestore

db = firestore.Client()

# ── Instruction 쿼리 ──

# 1) 내 워크스페이스 목록
my_workspaces = db.collection(
    "instructions/user:hong@sh-bank.com/workspaces"
).stream()

# 2) 특정 워크스페이스의 Instruction 목록
my_instructions = db.collection(
    "instructions/user:hong@sh-bank.com/workspaces/내_실험/items"
).stream()

# 3) 글로벌 Instruction 전체
global_instructions = db.collection(
    "instructions/_global/workspaces/표준_상담/items"
).stream()

# 4) 부서 Instruction 전체
dept_instructions = db.collection(
    "instructions/dept:IT팀/workspaces/IT_운영/items"
).stream()

# 5) 전체 Instruction 검색 (scope 무관)
all_instructions = db.collection_group("items").stream()

# 6) 전체 Semantic View 검색 (scope 무관)
all_views = db.collection_group("views").stream()

# 7) 특정 태그가 포함된 Instruction 검색
tagged = db.collection_group("items") \
    .where("tags", "array_contains", "상담") \
    .stream()

# 8) 특정 사용자가 만든 모든 Instruction (scope 무관)
user_items = db.collection_group("items") \
    .where("owner_id", "==", "hong@sh-bank.com") \
    .stream()

def get_visible_instructions(user_email: str, dept_id: str) -> list:
    """사용자가 접근 가능한 모든 Instruction을 반환"""
    sources = [
        f"instructions/_global/workspaces",
        f"instructions/dept:{dept_id}/workspaces",
        f"instructions/user:{user_email}/workspaces",
    ]

    results = []
    for source in sources:
        workspaces = db.collection(source).stream()
        for ws in workspaces:
            items = db.collection(f"{source}/{ws.id}/items").stream()
            results.extend(items)

    return results


visible = get_visible_instructions("hong@sh-bank.com", "IT팀")

def resolve_instruction(agent_id: str, user_email: str, dept_id: str) -> dict:
    """우선순위에 따라 에이전트의 Instruction을 해석"""

    # 1. 사용자별 배정 확인
    user_assignment = db.document(
        f"agent_assignments/user:{user_email}/agents/{agent_id}"
    ).get()

    if user_assignment.exists:
        ref = user_assignment.to_dict().get("instruction_ref")
        if ref:
            doc = db.document(ref).get()
            if doc.exists:
                return doc.to_dict()

    # 2. 부서 기본값 확인
    dept_assignment = db.document(
        f"agent_assignments/dept:{dept_id}/agents/{agent_id}"
    ).get()

    if dept_assignment.exists:
        ref = dept_assignment.to_dict().get("instruction_ref")
        if ref:
            doc = db.document(ref).get()
            if doc.exists:
                return doc.to_dict()

    # 3. 글로벌 기본값
    global_assignment = db.document(
        f"agent_assignments/_global/agents/{agent_id}"
    ).get()

    if global_assignment.exists:
        ref = global_assignment.to_dict().get("instruction_ref")
        if ref:
            doc = db.document(ref).get()
            if doc.exists:
                return doc.to_dict()

    return None

def assign_instruction_to_agent(
    user_email: str,
    agent_id: str,
    instruction_path: str,
    semantic_view_paths: list[str] = None
):
    """사용자가 자신의 Instruction을 특정 에이전트에 배정"""
    doc_ref = db.document(
        f"agent_assignments/user:{user_email}/agents/{agent_id}"
    )

    data = {
        "agent_id": agent_id,
        "instruction_ref": instruction_path,
        "semantic_view_refs": semantic_view_paths or [],
        "assigned_by": user_email,
        "updated_at": firestore.SERVER_TIMESTAMP,
    }

    doc_ref.set(data, merge=True)


# 사용 예시
assign_instruction_to_agent(
    user_email="hong@sh-bank.com",
    agent_id="financial-advisor",
    instruction_path="instructions/user:hong@sh-bank.com/workspaces/상담_커스텀/items/VIP_상담_지시",
    semantic_view_paths=[
        "semantic_views/_global/workspaces/표준_KPI/views/SV_예금잔액",
        "semantic_views/user:hong@sh-bank.com/workspaces/예금분석/views/SV_예금추이",
    ]
)

def promote_instruction(
    source_path: str,
    target_scope: str,
    target_workspace: str,
    promoted_by: str
):
    """Instruction을 상위 scope로 승격(복사)"""
    source_doc = db.document(source_path).get()
    if not source_doc.exists:
        raise ValueError(f"소스 문서를 찾을 수 없습니다: {source_path}")

    data = source_doc.to_dict()
    item_id = source_path.split("/")[-1]

    target_path = f"instructions/{target_scope}/workspaces/{target_workspace}/items/{item_id}"

    data.update({
        "promoted_from": source_path,
        "promoted_by": promoted_by,
        "scope": "global" if target_scope == "_global" else "dept",
        "visibility": "published",
        "promoted_at": firestore.SERVER_TIMESTAMP,
    })

    db.document(target_path).set(data)

    # 원본에 승격 이력 기록
    db.document(source_path).update({
        "promoted_to": target_path,
        "promoted_at": firestore.SERVER_TIMESTAMP,
    })

    return target_path


# 사용 예시: 개인 → 부서로 승격
promote_instruction(
    source_path="instructions/user:hong@sh-bank.com/workspaces/내_실험/items/VIP_상담_지시",
    target_scope="dept:IT팀",
    target_workspace="공유_지시",
    promoted_by="hong@sh-bank.com"
)