""" Motor de Derivações — Domínio CÓDIGO PYTHON PT-BR (Fase 3.2, seção 2.2 do plano) Segue a filosofia de scripts/derivacoes_gerar.py: conhecimento estruturado REAL (tarefa + enunciado + solução verificada + asserts executáveis) deriva deterministicamente em pares SFT múltiplos — sem professor, sem alucinação de fonte. Fontes (anti-mock): - sovereign_llm_engine/benchmarks/python_coding_evaluator.py (PY-01..PY-04) - data/codigo/tarefas_expandidas.json (PY-05..PY-15) GARANTIA: cada tarefa só entra na derivação se a SOLUÇÃO DE REFERÊNCIA passar nos asserts executáveis da própria tarefa (verificação real por subprocess). Se alguma falhar, o script ABORTA com erro — não emite dado não verificado. Saída: JSONL {id, tipo, origem, pergunta, resposta} Uso: python3 scripts/reft_dataset_codigo.py [--out data/sft/reft_codigo_v1.jsonl] """ import argparse import json import os import subprocess import sys import tempfile ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) sys.path.insert(0, ROOT_DIR) from sovereign_llm_engine.benchmarks.python_coding_evaluator import PYTHON_CODING_TASKS # --- Soluções de referência REAIS (verificadas contra os asserts antes de emitir) --- SOLUCOES = { "somar": ( "def somar(a, b):\n return a + b", [("pergunta", "Quanto é somar(2, 3)?", "somar(2, 3) = 5"), ("pergunta", "Quanto é somar(-1, 1)?", "somar(-1, 1) = 0"), ("inversa", "Qual chamada de somar retorna 5?", "somar(2, 3)")], ), "eh_par": ( "def eh_par(n):\n return n % 2 == 0", [("pergunta", "eh_par(4) retorna True ou False?", "eh_par(4) retorna True."), ("pergunta", "eh_par(7) retorna True ou False?", "eh_par(7) retorna False."), ("inversa", "Qual número eu passo para eh_par para obter True, usando um número menor que 10?", "eh_par(4) (também serviriam 0, 2, 6 ou 8)")], ), "inverter_texto": ( "def inverter_texto(s):\n return s[::-1]", [("pergunta", "O que retorna inverter_texto('brasil')?", "inverter_texto('brasil') retorna 'lisarb'."), ("pergunta", "O que retorna inverter_texto('')?", "inverter_texto('') retorna '' (string vazia)."), ("inversa", "Qual string invertida resulta em 'lisarb'?", "inverter_texto('brasil')")], ), "fatorial": ( "def fatorial(n):\n" " resultado = 1\n" " for i in range(2, n + 1):\n" " resultado *= i\n" " return resultado", [("pergunta", "Quanto é fatorial(5)?", "fatorial(5) = 120"), ("pergunta", "Quanto é fatorial(0)?", "fatorial(0) = 1 (por definição)."), ("inversa", "Qual chamada de fatorial retorna 120?", "fatorial(5)")], ), "maior_de_tres": ( "def maior_de_tres(a, b, c):\n return max(a, b, c)", [("pergunta", "Quanto é maior_de_tres(10, 5, 8)?", "maior_de_tres(10, 5, 8) = 10"), ("pergunta", "Quanto é maior_de_tres(-1, -5, -3)?", "maior_de_tres(-1, -5, -3) = -1"), ("inversa", "Que argumentos fazem maior_de_tres retornar 10?", "maior_de_tres(10, 5, 8)")], ), "eh_palindromo": ( "def eh_palindromo(s):\n return s == s[::-1]", [("pergunta", "eh_palindromo('arara') retorna True ou False?", "eh_palindromo('arara') retorna True."), ("pergunta", "eh_palindromo('casa') retorna True ou False?", "eh_palindromo('casa') retorna False."), ("inversa", "Qual palavra eu passo para eh_palindromo para obter True?", "eh_palindromo('arara')")], ), "contar_vogais": ( "def contar_vogais(s):\n" " total = 0\n" " for ch in s.lower():\n" " if ch in 'aeiou':\n" " total += 1\n" " return total", [("pergunta", "Quanto é contar_vogais('banana')?", "contar_vogais('banana') = 3"), ("pergunta", "Quanto é contar_vogais('abacaxi')?", "contar_vogais('abacaxi') = 4"), ("inversa", "Que palavra dá contar_vogais igual a 3?", "contar_vogais('banana') = 3")], ), "media": ( "def media(numeros):\n return sum(numeros) / len(numeros)", [("pergunta", "Quanto é media([2, 4, 6])?", "media([2, 4, 6]) = 4"), ("pergunta", "Quanto é media([10])?", "media([10]) = 10"), ("inversa", "Qual lista dá media igual a 4?", "media([2, 4, 6]) = 4")], ), "fibonacci": ( "def fibonacci(n):\n" " a, b = 0, 1\n" " for _ in range(n):\n" " a, b = b, a + b\n" " return a", [("pergunta", "Quanto é fibonacci(7)?", "fibonacci(7) = 13"), ("pergunta", "Quanto é fibonacci(10)?", "fibonacci(10) = 55"), ("pergunta", "Quanto é fibonacci(0)?", "fibonacci(0) = 0"), ("inversa", "Qual n faz fibonacci(n) retornar 55?", "fibonacci(10) = 55")], ), "celsius_para_fahrenheit": ( "def celsius_para_fahrenheit(c):\n return c * 9 / 5 + 32", [("pergunta", "Quanto é celsius_para_fahrenheit(100)?", "celsius_para_fahrenheit(100) = 212"), ("pergunta", "Quanto é celsius_para_fahrenheit(-40)?", "celsius_para_fahrenheit(-40) = -40"), ("inversa", "Qual temperatura Celsius dá 212 Fahrenheit?", "celsius_para_fahrenheit(100) = 212")], ), "contar_palavras": ( "def contar_palavras(frase):\n return len(frase.split())", [("pergunta", "Quanto é contar_palavras('a b c d')?", "contar_palavras('a b c d') = 4"), ("pergunta", "Quanto é contar_palavras('olá mundo')?", "contar_palavras('olá mundo') = 2"), ("inversa", "Que frase dá contar_palavras igual a 4?", "contar_palavras('a b c d') = 4")], ), "remover_duplicados": ( "def remover_duplicados(lista):\n" " vistos = set()\n" " resultado = []\n" " for item in lista:\n" " if item not in vistos:\n" " vistos.add(item)\n" " resultado.append(item)\n" " return resultado", [("pergunta", "O que retorna remover_duplicados([1, 2, 2, 3])?", "remover_duplicados([1, 2, 2, 3]) retorna [1, 2, 3]."), ("pergunta", "remover_duplicados preserva a ordem original?", "Sim, remover_duplicados preserva a ordem original dos elementos."), ("inversa", "Que lista dá remover_duplicados igual a ['a', 'b']?", "remover_duplicados(['a', 'a', 'b']) = ['a', 'b']")], ), "eh_primo": ( "def eh_primo(n):\n" " if n < 2:\n" " return False\n" " i = 2\n" " while i * i <= n:\n" " if n % i == 0:\n" " return False\n" " i += 1\n" " return True", [("pergunta", "eh_primo(17) retorna True ou False?", "eh_primo(17) retorna True."), ("pergunta", "eh_primo(15) retorna True ou False?", "eh_primo(15) retorna False (15 = 3 × 5)."), ("pergunta", "eh_primo(1) retorna True ou False?", "eh_primo(1) retorna False (1 não é primo por definição)."), ("inversa", "Qual número menor que 20 dá eh_primo igual a True?", "eh_primo(2), eh_primo(3), eh_primo(5), etc.")], ), "segundo_maior": ( "def segundo_maior(numeros):\n" " unicos = sorted(set(numeros))\n" " return unicos[-2]", [("pergunta", "Quanto é segundo_maior([1, 5, 3])?", "segundo_maior([1, 5, 3]) = 3"), ("pergunta", "Quanto é segundo_maior([7, 7, 4, 9])?", "segundo_maior([7, 7, 4, 9]) = 7"), ("inversa", "Que lista dá segundo_maior igual a 3?", "segundo_maior([1, 5, 3]) = 3")], ), "somar_digitos": ( "def somar_digitos(n):\n" " total = 0\n" " while n > 0:\n" " total += n % 10\n" " n //= 10\n" " return total", [("pergunta", "Quanto é somar_digitos(123)?", "somar_digitos(123) = 6"), ("pergunta", "Quanto é somar_digitos(99999)?", "somar_digitos(99999) = 45"), ("inversa", "Qual número tem somar_digitos igual a 6?", "somar_digitos(123) = 6 (também 1230, 321…)")], ), } # Paráfrases de pedido direto (todas reais, mesmas instruções reformuladas) PARAFRASES = [ "Crie uma função Python chamada `{fn}` que {descricao_curta}.", "Implemente em Python a função `{fn}`, que {descricao_curta}.", "Preciso de uma função Python: `{fn}`. Ela deve {descricao_curta}.", "Escreva o código da função `{fn}` em Python, que {descricao_curta}.", "Em Python, como eu escrevo uma função `{fn}` que {descricao_curta}?", "Me dê o código Python de uma função `{fn}` que {descricao_curta}.", "Função Python `{fn}`: {descricao_curta}. Escreva ela.", ] # Implementações alternativas REAIS por função (cada uma verificada contra os asserts) ALTERNATIVAS = { "somar": ["def somar(a, b):\n return sum([a, b])"], "eh_par": ["def eh_par(n):\n return not n % 2", "def eh_par(n):\n return n & 1 == 0"], "inverter_texto": ["def inverter_texto(s):\n return ''.join(reversed(s))"], "fatorial": ["def fatorial(n):\n" " if n <= 1:\n" " return 1\n" " return n * fatorial(n - 1)", "import math\n\n" "def fatorial(n):\n return math.prod(range(1, n + 1))"], "maior_de_tres": ["def maior_de_tres(a, b, c):\n" " if a >= b and a >= c:\n" " return a\n" " if b >= c:\n" " return b\n" " return c"], "eh_palindromo": ["def eh_palindromo(s):\n" " invertida = ''\n" " for ch in s:\n" " invertida = ch + invertida\n" " return s == invertida"], "contar_vogais": ["def contar_vogais(s):\n" " return sum(1 for ch in s.lower() if ch in 'aeiou')"], "media": ["import statistics\n\n" "def media(numeros):\n return statistics.mean(numeros)", "def media(numeros):\n" " total = 0\n" " for n in numeros:\n" " total += n\n" " return total / len(numeros)"], "fibonacci": ["def fibonacci(n):\n" " if n < 2:\n" " return n\n" " return fibonacci(n - 1) + fibonacci(n - 2)", "def fibonacci(n):\n" " seq = [0, 1]\n" " for i in range(2, n + 1):\n" " seq.append(seq[i - 1] + seq[i - 2])\n" " return seq[n]"], "celsius_para_fahrenheit": ["def celsius_para_fahrenheit(c):\n" " return c * 1.8 + 32"], "contar_palavras": ["def contar_palavras(frase):\n" " palavras = frase.split(' ')\n" " return len(palavras)"], "remover_duplicados": ["def remover_duplicados(lista):\n" " resultado = []\n" " for item in lista:\n" " if item not in resultado:\n" " resultado.append(item)\n" " return resultado", "def remover_duplicados(lista):\n" " return list(dict.fromkeys(lista))"], "eh_primo": ["def eh_primo(n):\n" " if n < 2:\n" " return False\n" " for i in range(2, int(n ** 0.5) + 1):\n" " if n % i == 0:\n" " return False\n" " else:\n" " return True"], "segundo_maior": ["def segundo_maior(numeros):\n" " unicos = list(set(numeros))\n" " unicos.remove(max(unicos))\n" " return max(unicos)", "def segundo_maior(numeros):\n" " maior = segundo = float('-inf')\n" " for n in numeros:\n" " if n > maior:\n" " segundo, maior = maior, n\n" " elif n > segundo:\n" " segundo = n\n" " return segundo"], "somar_digitos": ["def somar_digitos(n):\n" " return sum(int(digito) for digito in str(n))"], } TEMPLATES_ALT = [ "Escreva uma função Python chamada `{fn}` que {descricao_curta}. Use uma abordagem diferente do óbvio.", "Resolva em Python: `{fn}` deve {descricao_curta}.", ] # Reconhecimento e assinatura TPL_ASSINATURA = ("Qual é a assinatura de uma função Python que {descricao_curta}? " "Responda apenas com a linha `def`.") TPL_RECONHECIMENTO = "Que nome de função adequado para uma função Python que {descricao_curta}?" TPL_COMO_FAZO = "Em Python, como faço para {descricao_curta_infinitivo}?" DESCRICAO_CURTA = { "somar": "receba dois números e retorne a soma deles", "eh_par": "retorne True se o número for par e False caso contrário", "inverter_texto": "receba uma string e retorne o texto invertido", "fatorial": "calcule o fatorial de um número inteiro não-negativo", "maior_de_tres": "retorne o maior dos três números recebidos", "eh_palindromo": "retorne True se a string lê-se igual de trás para frente e False caso contrário", "contar_vogais": "retorne quantas vogais existem na string", "media": "receba uma lista de números e retorne a média aritmética", "fibonacci": "retorne o n-ésimo termo da sequência de Fibonacci (fibonacci(0) = 0, fibonacci(1) = 1)", "celsius_para_fahrenheit": "converta graus Celsius para Fahrenheit (F = C * 9/5 + 32)", "contar_palavras": "retorne o número de palavras separadas por espaços", "remover_duplicados": "retorne uma nova lista sem elementos duplicados, preservando a ordem original", "eh_primo": "retorne True se o número inteiro positivo for primo e False caso contrário", "segundo_maior": "retorne o segundo maior valor de uma lista de números distintos", "somar_digitos": "retorne a soma dos dígitos de um número inteiro não-negativo", } # Mesmas descrições em infinitivo (para "como faço para ...") DESCRICAO_INFINITIVO = { "somar": "somar dois números", "eh_par": "verificar se um número é par", "inverter_texto": "inverter uma string", "fatorial": "calcular o fatorial de um número", "maior_de_tres": "achar o maior de três números", "eh_palindromo": "verificar se uma palavra é palíndromo", "contar_vogais": "contar as vogais de uma string", "media": "calcular a média de uma lista de números", "fibonacci": "obter o n-ésimo termo de Fibonacci", "celsius_para_fahrenheit": "converter Celsius para Fahrenheit", "contar_palavras": "contar as palavras de uma frase", "remover_duplicados": "remover duplicados de uma lista mantendo a ordem", "eh_primo": "verificar se um número é primo", "segundo_maior": "achar o segundo maior de uma lista", "somar_digitos": "somar os dígitos de um número", } def verificar_solucao(funcao, test_code): """Executa solução + asserts em subprocess isolado. Retorna (ok, detalhe).""" full = funcao + "\n\n" + test_code + "\nprint('__OK__')\n" with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as f: f.write(full) tmp = f.name try: proc = subprocess.run([sys.executable, tmp], capture_output=True, text=True, timeout=10, cwd=tempfile.gettempdir()) ok = proc.returncode == 0 and "__OK__" in proc.stdout return ok, (proc.stdout + "|ERR|" + proc.stderr)[-300:] if not ok else "asserts OK" except Exception as e: return False, f"{type(e).__name__}: {e}" finally: os.unlink(tmp) import ast def parsear_asserts(test_code): """Extrai (chamada_str, resultado_str) de cada assert executável real do banco.""" pares = [] for linha in test_code.splitlines(): linha = linha.strip() if not linha.startswith("assert "): continue expr = linha[len("assert "):].strip() try: arvore = ast.parse(expr, mode="eval").body except SyntaxError: continue if isinstance(arvore, ast.Compare) and isinstance(arvore.left, ast.Call): chamada = ast.unparse(arvore.left) if len(arvore.ops) == 1 and isinstance(arvore.ops[0], ast.Eq): resultado = ast.unparse(arvore.comparators[0]) pares.append((chamada, resultado)) elif isinstance(arvore, ast.Call): # assert fn(...) is True/False resto = expr[expr.rfind(")") + 1:].strip() m = resto.startswith("is ") if m: pares.append((ast.unparse(arvore), resto[3:].strip())) return pares def carregar_tarefas(): """Une PY-01..04 (embutidas) com PY-05..15 (banco expandido).""" tarefas = list(PYTHON_CODING_TASKS) caminho = os.path.join(ROOT_DIR, "data", "codigo", "tarefas_expandidas.json") with open(caminho, encoding="utf-8") as f: banco = json.load(f) ids_existentes = {t["id"] for t in tarefas} for t in banco["tarefas"]: if t["id"] not in ids_existentes: tarefas.append(t) tarefas.sort(key=lambda t: t["id"]) return tarefas def derivar_tarefas(tarefas): pares = [] for t in tarefas: fn, pid = t["function_name"], t["id"] if fn not in SOLUCOES: print(f"[AVISO] Sem solução de referência para {fn} ({pid}) — tarefa pulada") continue solucao, qas = SOLUCOES[fn] ok, detalhe = verificar_solucao(solucao, t["test_code"]) if not ok: sys.exit(f"[ERRO ANTI-MOCK] Solução de {fn} FALHOU nos asserts da própria tarefa:\n{detalhe}") desc = DESCRICAO_CURTA[fn] desc_inf = DESCRICAO_INFINITIVO[fn] codigo_md = f"```python\n{solucao}\n```" # T1 — direto (enunciado oficial → código) pares.append({"tipo": "direto", "origem": pid, "pergunta": t["prompt"], "resposta": codigo_md}) # T2 — paráfrases for tpl in PARAFRASES: pares.append({"tipo": "parafrase", "origem": pid, "pergunta": tpl.format(fn=fn, descricao_curta=desc), "resposta": codigo_md}) # T3 — explicação + código explicacao = (f"A função `{fn}` resolve o problema aplicando diretamente a regra " f"do enunciado sobre os dados de entrada. Implementação:\n{codigo_md}") pares.append({"tipo": "explicacao", "origem": pid, "pergunta": f"Explique como resolver: {t['prompt']}", "resposta": explicacao}) # T4 — QA manual curado (comportamento real, derivado dos asserts) for modo, pergunta, resposta in qas: pares.append({"tipo": f"qa_{modo}", "origem": pid, "pergunta": pergunta, "resposta": resposta}) # T4b — QA automático de TODOS os asserts (chamada → resultado) vistos_qa = {q.lower() for _, q, _ in qas} chamadas_assert = parsear_asserts(t["test_code"]) for chamada, resultado in chamadas_assert: pergunta = f"O que retorna {chamada}?" if pergunta.lower() in vistos_qa: continue vistos_qa.add(pergunta.lower()) if resultado in ("True", "False"): resposta = f"{chamada} retorna {resultado}." else: resposta = f"{chamada} = {resultado}" pares.append({"tipo": "qa_auto", "origem": pid, "pergunta": pergunta, "resposta": resposta}) # T4c — inversas automáticas não-ambíguas (saída única → entrada canônica) saidas_vistas = {} for chamada, resultado in chamadas_assert: saidas_vistas.setdefault(resultado, []).append(chamada) for resultado, chamadas in sorted(saidas_vistas.items()): if len(chamadas) != 1: continue # ambíguo — não deriva inversa if resultado in ("True", "False"): continue # booleano é sempre ambíguo semanticamente chamada = chamadas[0] pares.append({"tipo": "qa_inversa_auto", "origem": pid, "pergunta": f"Qual chamada de `{fn}` retorna {resultado}?", "resposta": chamada}) # T5 — docstring pares.append({"tipo": "docstring", "origem": pid, "pergunta": f"Escreva a função `{fn}` com uma docstring explicando o que ela faz.", "resposta": codigo_md.replace("```python\n", "```python\n" + f'"""{t["task_name"]}."""\n', 1)}) # T6 — teste unitário pares.append({"tipo": "testes", "origem": pid, "pergunta": f"Mostre testes de asserção para validar a função `{fn}`.", "resposta": f"```python\n{t['test_code']}\n```"}) # T7 — implementações alternativas REAIS, cada uma verificada contra os asserts for alt in ALTERNATIVAS.get(fn, []): ok_alt, detalhe_alt = verificar_solucao(alt, t["test_code"]) if not ok_alt: sys.exit(f"[ERRO ANTI-MOCK] Alternativa de {fn} FALHOU nos asserts:\n{detalhe_alt}") alt_md = f"```python\n{alt}\n```" tpl = TEMPLATES_ALT[len(pares) % len(TEMPLATES_ALT)] pares.append({"tipo": "alternativa", "origem": pid, "pergunta": tpl.format(fn=fn, descricao_curta=desc), "resposta": alt_md}) pares.append({"tipo": "alternativa_explicada", "origem": pid, "pergunta": f"Resolva com uma implementação enxuta: {desc[0].upper()}{desc[1:]} " f"(função `{fn}`).", "resposta": f"Uma forma direta:\n{alt_md}"}) # T8 — assinatura / reconhecimento / como-faço import re as _re m_def = _re.match(r"def (\w+)\(([^)]*)\)", solucao.splitlines()[0]) linha_def = f"def {m_def.group(1)}({m_def.group(2)}):" pares.append({"tipo": "assinatura", "origem": pid, "pergunta": TPL_ASSINATURA.format(descricao_curta=desc), "resposta": f"`{linha_def}`"}) pares.append({"tipo": "reconhecimento", "origem": pid, "pergunta": TPL_RECONHECIMENTO.format(descricao_curta=desc), "resposta": f"`{fn}`"}) pares.append({"tipo": "como_faco", "origem": pid, "pergunta": TPL_COMO_FAZO.format(descricao_curta_infinitivo=desc_inf), "resposta": codigo_md}) # T9 — código cru (sem markdown; o extrator do avaliador também lê assim) pares.append({"tipo": "codigo_cru", "origem": pid, "pergunta": f"Responda apenas com código: {desc[0].upper()}{desc[1:]} " f"(função `{fn}`).", "resposta": solucao}) # T10 — verificação de teste real (primeiro assert do banco) if chamadas_assert: chamada, resultado = chamadas_assert[0] pares.append({"tipo": "verificacao_teste", "origem": pid, "pergunta": f"O teste `assert {chamada} == {resultado}` passa? Por quê?" if resultado not in ("True", "False") else f"O teste `assert {chamada} is {resultado}` passa? Por quê?", "resposta": f"Passa. Porque a função correta satisfaz {chamada} = {resultado}. " f"Implementação:\n{codigo_md}"}) return pares def main(): ap = argparse.ArgumentParser() ap.add_argument("--out", default=os.path.join(ROOT_DIR, "data", "sft", "reft_codigo_v1.jsonl")) args = ap.parse_args() tarefas = carregar_tarefas() print(f"Tarefas carregadas: {len(tarefas)} ({tarefas[0]['id']}..{tarefas[-1]['id']})") pares = derivar_tarefas(tarefas) os.makedirs(os.path.dirname(args.out), exist_ok=True) with open(args.out, "w", encoding="utf-8") as f: for p in pares: f.write(json.dumps(p, ensure_ascii=False) + "\n") tipos = {} for p in pares: tipos[p["tipo"]] = tipos.get(p["tipo"], 0) + 1 print(f"\nDataset: {args.out}") print(f"Total de pares SFT: {len(pares)}") for k, v in sorted(tipos.items()): print(f" • {k:<14} {v}") if __name__ == "__main__": main()