ruicesar/instrucoes_gerais
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: switch agent executor to a zero-dependency Gemini 3 Flash multi-turn tool engine

Browse Source
This commit is contained in:
Rui
2026-05-17 03:07:38 +02:00
parent d92f06fd7f
commit f50dc35be1
1 changed files with 198 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
templates/agent_executor.py
+197 -88
View File
@@ -1,6 +1,11 @@
import os, json, redis, subprocess, base64
from urllib import request, parse
from anthropic import Anthropic
import os
import json
import redis
import subprocess
import base64
import random
import urllib.request
import urllib.error
class SwarmAgent:
def __init__(self, task_data):
@@ -9,46 +14,34 @@ class SwarmAgent:
self.repo_url = f"http://10.99.0.3:3000/{self.repo_name}.git"
self.local_path = os.path.join("/home/deploy/projetos", self.repo_name.split('/')[-1])
# Conexão Redis para pegar chaves
# Conexão Redis para pegar chaves do pool
self.r = redis.Redis(host='10.99.0.3', port=6379, password='clube67_dragonfly_pass_9903', db=0, decode_responses=True)
def get_api_key(self):
def get_api_key(self, provider):
pool = json.loads(self.r.get('swarm:keys:pool'))
return random.choice([k['key'] for k in pool['anthropic'] if k['status'] == 'active'])
active_keys = [k['key'] for k in pool.get(provider, []) if k.get('status') == 'active']
if not active_keys:
raise Exception(f"Nenhuma chave ativa encontrada para o provedor: {provider}")
return random.choice(active_keys)
def setup_workspace(self):
os.makedirs("/home/deploy/projetos", exist_ok=True)
if not os.path.exists(self.local_path):
print(f" [>] Clonando repositório {self.repo_url} em {self.local_path}...")
subprocess.run(["git", "clone", self.repo_url, self.local_path])
else:
print(f" [>] Atualizando repositório em {self.local_path}...")
subprocess.run(["git", "-C", self.local_path, "pull"])
def run_bash(self, command):
"""Ferramenta para a IA rodar comandos na VPS 4"""
result = subprocess.run(command, shell=True, capture_output=True, text=True, cwd=self.local_path)
return f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}"
def post_comment(self, message):
issue_id = self.data.get('data', {}).get('issue', {}).get('number')
if not issue_id: return
url = f"http://10.99.0.3:3000/api/v1/repos/{self.repo_name}/issues/{issue_id}/comments"
auth_str = 'ruicesar:h$tg@g5aga$ra1E3$C-yHW$-BA@DF2@Grfa!3#'
base64_auth = base64.b64encode(auth_str.encode('ascii')).decode('ascii')
req = request.Request(url, data=json.dumps({"body": message}).encode('utf-8'), method='POST')
req.add_header('Content-Type', 'application/json')
req.add_header('Authorization', f'Basic {base64_auth}')
try:
with request.urlopen(req) as res: pass
except: pass
def run_command(self, command):
"""Executa comandos shell na VPS 4"""
print(f" [Exec] Rodando: {command}")
print(f" [Exec] Rodando comando: {command}")
result = subprocess.run(command, shell=True, capture_output=True, text=True, cwd=self.local_path)
return {"stdout": result.stdout, "stderr": result.stderr}
def write_file(self, path, content):
"""Escreve/Edita arquivos no projeto"""
print(f" [Exec] Escrevendo arquivo: {path}")
full_path = os.path.join(self.local_path, path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)
with open(full_path, "w") as f:
@@ -57,11 +50,91 @@ class SwarmAgent:
def read_file(self, path):
"""Lê arquivos do projeto"""
print(f" [Exec] Lendo arquivo: {path}")
full_path = os.path.join(self.local_path, path)
if os.path.exists(full_path):
with open(full_path, "r") as f:
return f.read()
return "Erro: Arquivo não encontrado."
return f"Erro: Arquivo {path} não encontrado."
def post_comment(self, message):
"""Posta um comentário de feedback na Issue do Gitea"""
issue_id = self.data.get('data', {}).get('issue', {}).get('number')
if not issue_id:
return
url = f"http://10.99.0.3:3000/api/v1/repos/{self.repo_name}/issues/{issue_id}/comments"
# Auth Basic
auth_str = 'ruicesar:h$tg@g5aga$ra1E3$C-yHW$-BA@DF2@Grfa!3#'
base64_auth = base64.b64encode(auth_str.encode('ascii')).decode('ascii')
data = json.dumps({"body": message}).encode('utf-8')
req = urllib.request.Request(url, data=data, method='POST')
req.add_header('Content-Type', 'application/json')
req.add_header('Authorization', f'Basic {base64_auth}')
try:
with urllib.request.urlopen(req) as response:
if response.status == 201:
print(f" [✓] Comentário postado na Issue #{issue_id}")
except Exception as e:
print(f" [E] Falha ao postar comentário no Gitea: {e}")
def call_gemini(self, contents, system_instruction, api_key):
"""Chama a API do Gemini com ferramentas ativas em formato puramente nativo"""
url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-flash-latest:generateContent?key={api_key}"
headers = {'content-type': 'application/json'}
tools = [{
"functionDeclarations": [
{
"name": "read_file",
"description": "Lê o conteúdo de um arquivo do projeto para análise técnica.",
"parameters": {
"type": "OBJECT",
"properties": {
"path": {"type": "STRING", "description": "Caminho relativo do arquivo no repositório."}
},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "Cria ou edita um arquivo com o novo conteúdo fornecido.",
"parameters": {
"type": "OBJECT",
"properties": {
"path": {"type": "STRING", "description": "Caminho relativo do arquivo no repositório."},
"content": {"type": "STRING", "description": "Conteúdo completo a ser gravado no arquivo."}
},
"required": ["path", "content"]
}
},
{
"name": "run_command",
"description": "Executa um comando de terminal (como NPM, Docker, Git, etc) no diretório do projeto.",
"parameters": {
"type": "OBJECT",
"properties": {
"command": {"type": "STRING", "description": "Comando shell a ser executado."}
},
"required": ["command"]
}
}
]
}]
payload = {
"contents": contents,
"systemInstruction": {
"parts": [{"text": system_instruction}]
},
"tools": tools
}
req = urllib.request.Request(url, data=json.dumps(payload).encode('utf-8'), headers=headers, method='POST')
with urllib.request.urlopen(req) as res:
return json.loads(res.read().decode('utf-8'))
def execute(self):
self.setup_workspace()
@@ -70,80 +143,116 @@ class SwarmAgent:
vps_ip = subprocess.getoutput("hostname -I").split()[0]
vps_name = subprocess.getoutput("hostname")
with open("/home/deploy/instrucoes/global/doutrina_agentes.md", "r") as f: doutrina = f.read()
# Carrega a Doutrina de Agentes
doutrina_path = "/home/deploy/instrucoes/global/doutrina_agentes.md"
doutrina = ""
if os.path.exists(doutrina_path):
with open(doutrina_path, "r") as f:
doutrina = f.read()
# Consciência Situacional Injetada
situational_awareness = f"\n\n[CONSCIÊNCIA SITUACIONAL]: Você está rodando na VPS: {vps_name} (IP: {vps_ip})\n"
system_instruction = doutrina + situational_awareness
api_key = self.get_api_key()
client = Anthropic(api_key=api_key)
task = self.data.get('data', {}).get('issue', {}).get('body', 'Corrigir projeto')
print(f" [!] Agente iniciando na {vps_name} ({vps_ip}) para resolver a tarefa.")
print(f" [!] Agente iniciando na {vps_name} ({vps_ip})...")
task = self.data.get('data', {}).get('issue', {}).get('body', 'Sem descrição')
# O Agente agora tem um loop de pensamento e ferramentas
# Para este MVP, vamos permitir que ele faça uma análise e execute uma rodada de ferramentas
tools = [
# Histórico inicial
contents = [
{
"name": "read_file",
"description": "Lê o conteúdo de um arquivo do projeto",
"input_schema": {
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "Escreve ou edita um arquivo no projeto",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"}
},
"required": ["path", "content"]
}
},
{
"name": "run_command",
"description": "Executa um comando shell (ex: docker compose up)",
"input_schema": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"]
}
"role": "user",
"parts": [{"text": f"Tarefa: {task}. Diretório do projeto: {self.local_path}. Analise o código e corrija a regressão aplicando as mudanças necessárias."}]
}
]
response = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=4000,
system=doutrina + situational_awareness,
tools=tools,
messages=[{"role": "user", "content": f"Tarefa: {task}\nAnalise o diretório {self.local_path} e resolva o problema."}]
api_key = self.get_api_key('google')
feedback_steps = []
final_text = ""
# Loop de multi-turn de ferramentas (máximo 7 iterações)
for turn in range(7):
print(f" [>] Iniciando turno {turn + 1} com o cérebro Gemini...")
try:
response = self.call_gemini(contents, system_instruction, api_key)
except Exception as e:
print(f" [E] Erro ao chamar a API do Gemini: {e}")
feedback_steps.append(f"* **Erro na API**: {e}")
break
candidates = response.get('candidates', [])
if not candidates:
print(" [E] Nenhuma resposta retornada.")
break
content = candidates[0].get('content', {})
parts = content.get('parts', [])
# Mantém no histórico a resposta do modelo
contents.append(content)
has_tool_call = False
function_responses = []
for part in parts:
if 'text' in part:
final_text += part['text'] + "\n"
if 'functionCall' in part:
has_tool_call = True
func_call = part['functionCall']
func_name = func_call.get('name')
args = func_call.get('args', {})
print(f" [!] Executando ferramenta: {func_name} (args: {args})")
result_data = None
try:
if func_name == "read_file":
path = args.get('path')
result_data = self.read_file(path)
feedback_steps.append(f"* **Lendo**: `{path}`")
elif func_name == "write_file":
path = args.get('path')
content_to_write = args.get('content')
result_data = self.write_file(path, content_to_write)
feedback_steps.append(f"* **Escrevendo**: `{path}`")
elif func_name == "run_command":
cmd = args.get('command')
result_data = self.run_command(cmd)
feedback_steps.append(f"* **Rodando**: `{cmd}`")
except Exception as e_func:
print(f" [E] Erro ao rodar ferramenta {func_name}: {e_func}")
result_data = f"Erro de execução da ferramenta: {e_func}"
function_responses.append({
"functionResponse": {
"name": func_name,
"response": {
"output": result_data
}
}
})
if not has_tool_call:
print(" [🏁] Ciclo finalizado pelo modelo de IA.")
break
else:
contents.append({
"role": "function",
"parts": function_responses
})
# Formata o feedback e posta de volta no Gitea
steps_str = "\n".join(feedback_steps)
final_comment = (
f"### 🤖 [VPS 4 - EXECUTOR SWARM]:\n\n"
f"{final_text}\n\n"
f"**⚙️ Ações e Diagnóstico de Infraestrutura:**\n{steps_str}"
)
# Processamento simples de ferramentas (para este estágio)
final_feedback = f"### 🤖 [VPS 4 - EXECUTOR]:\n\n{response.content[0].text}\n\n"
for content in response.content:
if content.type == "tool_use":
if content.name == "read_file":
res = self.read_file(content.input["path"])
final_feedback += f"* **Lendo**: {content.input['path']}\n"
elif content.name == "write_file":
res = self.write_file(content.input["path"], content.input["content"])
final_feedback += f"* **Escrevendo**: {content.input['path']}\n"
elif content.name == "run_command":
res = self.run_command(content.input["command"])
final_feedback += f"* **Rodando**: `{content.input['command']}`\n"
self.post_comment(final_feedback)
self.post_comment(final_comment)
if __name__ == "__main__":
import argparse, random
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--data")
args = parser.parse_args()