Spaces:
Build error
Build error
File size: 3,922 Bytes
5870d1d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | # aduc_framework/engineers/composer.py
#
# Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos
#
# Versão 1.0.0 (Conversational Execution Engine)
import logging
from PIL import Image
from typing import List, Dict, Any, Generator
import json
from ..managers.llama_multimodal_manager import llama_multimodal_manager_singleton
logger = logging.getLogger(__name__)
# ... (função robust_json_parser aqui) ...
class Composer:
"""
O Composer atua como o Mestre de Obras. Ele recebe um plano de execução
e supervisiona o LLM para realizar cada tarefa, construindo o
artefato final (o DNA de pré-produção) de forma iterativa e conversacional.
"""
def __init__(self):
# O Composer agora carrega um "caderno de plantas" com os prompts para cada tarefa
self.prompts = {
"PREPROD_01_CATALOG_ASSETS": self._read_prompt_template("catalog_assets_prompt.txt"),
"PREPROD_02_SCORE_ASSETS": self._read_prompt_template("score_assets_prompt.txt"),
# ... e assim por diante para cada task_id do Deformes2D
}
def _read_prompt_template(self, filename: str) -> str:
# ... (implementação para ler os arquivos de prompt)
pass
def _talk_to_llama(self, prompt: str, images: List[Image.Image] = None, expected_format="text") -> Any:
# ... (a mesma função de wrapper de antes)
pass
# O PONTO DE ENTRADA PRINCIPAL DO COMPOSER
def execute_plan(
self,
execution_plan: List[Dict[str, Any]],
initial_data: Dict[str, Any]
) -> Generator[Dict[str, Any], None, Dict[str, Any]]:
"""
Executa um plano de trabalho tarefa por tarefa, emitindo atualizações
de progresso e construindo o DNA.
"""
llama_multimodal_manager_singleton.reset_chat()
# O DNA começa com os dados iniciais do Orquestrador
dna = {
"global_prompt": initial_data["global_prompt"],
"initial_media_paths": initial_data["user_media_paths"],
"asset_catalog": {},
"story_summary": "",
"scenes": []
}
user_media = [Image.open(p) for p in initial_data["user_media_paths"]]
# Loop principal de execução do plano
for i, task in enumerate(execution_plan):
task_id = task['task_id']
description = task['description']
# --- ATUALIZAÇÃO PARA A UI ---
# Emite o estado atual antes de executar a tarefa
yield {
"status": "progress",
"stage": f"{i+1}/{len(execution_plan)}",
"current_task": task_id,
"message": description,
"dna_snapshot": dna
}
# Executa a lógica específica para cada tarefa
if task_id == "PREPROD_01_CATALOG_ASSETS":
prompt = self.prompts[task_id].format(**task['inputs'])
asset_catalog = self._talk_to_llama(prompt, user_media, expected_format="json")
dna["asset_catalog"] = asset_catalog
elif task_id == "PREPROD_02_SCORE_ASSETS":
# A conversa continua, o Llama lembra do catálogo que acabou de criar
prompt = self.prompts[task_id].format(**task['inputs'])
scored_catalog = self._talk_to_llama(prompt, expected_format="json")
dna["asset_catalog"] = scored_catalog
# ... lógica para as outras tarefas (PREPROD_03, 04, etc.)
task['status'] = 'complete'
# --- FIM ---
final_message = "Plano de pré-produção concluído com sucesso."
logger.info(final_message)
yield {"status": "complete", "message": final_message, "dna": dna}
return dna
# Instância Singleton
composer_singleton = Composer() |