Владимир

import os import json import asyncio import aiohttp from openai import OpenAI # --- КОНФИГУРАЦИЯ --- MOLTJOBS_API_KEY = os.getenv("MOLTJOBS_API_KEY", "mj_live_your_key") MOLTJOBS_BASE_URL = "https://moltjobs.io" OPENAI_KEY = os.getenv("OPENAI_API_KEY", "your_openai_key") openai_client = OpenAI(api_key=OPENAI_KEY) class MoltJobsExpertAgent: def __init__(self, agent_id, specialties: list): self.agent_id = agent_id self.specialties = specialties # Например: ["data-analysis", "python-engineering"] self.headers = { "Authorization": f"Bearer {MOLTJOBS_API_KEY}", "Content-Type": "application/json" } # 1. МОНИТОРИНГ: Поиск сложных открытых задач async def fetch_open_jobs(self, session: aiohttp.ClientSession) -> list: """Сканирует платформу на наличие открытых задач""" url = f"{MOLTJOBS_BASE_URL}/jobs?status=OPEN" async with session.get(url, headers=self.headers) as response: if response.status == 200: jobs = await response.json() print(f"[🔍 Market] Discovered {len(jobs)} open jobs.") return jobs else: print(f"[❌ Error] Failed to fetch jobs: {response.status}") return [] # 2. ОЦЕНКА И ФИЛЬТРАЦИЯ: Выбор подходящих тасков def evaluate_job(self, job: dict) -> bool: """Анализирует сложность, бюджет и соответствие навыкам""" # Сложные задачи обычно требуют структурированных входных данных и имеют более высокий бюджет budget = job.get("budget", 0) requirements = job.get("spec", {}).get("acceptance_criteria", "") # Проверяем соответствие темы тегам агента matching_specialty = any(tag in job.get("tags", []) for tag in self.specialties) # Критерий «сложной и выгодной» задачи: бюджет > 100 USDC и соответствие специализации if budget >= 100 and matching_specialty: print(f"[🔥 Evaluation] High-value job matched! ID: {job['id']}, Budget: {budget} USDC") return True return False # 3. АВТО-СТАВКА (BID): Отправка предложения на биржу async def place_bid(self, session: aiohttp.ClientSession, job_id: str, budget: float) -> bool: """Подает автоматическую заявку на выполнение работы""" url = f"{MOLTJOBS_BASE_URL}/bids" payload = { "agent_id": self.agent_id, "job_id": job_id, "proposed_amount": budget, "proposal_rationale": "I am an advanced reasoning agent specialized in architectural tasks and zero-error execution." } async with session.post(url, headers=self.headers, json=payload) as response: if response.status in: res_data = await response.json() print(f"[🚀 Bid] Successfully placed bid for job {job_id}. Bid ID: {res_data.get('id')}") return True else: print(f"[⚠️ Bid] Bid rejected for job {job_id}: {response.status}") return False # 4. РАССУЖДЕНИЕ И ВЫПОЛНЕНИЕ: Ядро сложного решения (LLM ReAct Loop) def execute_complex_task(self, job_spec: dict) -> str: """Запускает глубокий цикл рассуждения ИИ для генерации артефакта работы""" print(f"[🧠 Reasoning Engine] Processing complex task criteria...") system_prompt = ( "You are an Elite Autonomous Operator on MoltJobs. Your goal is to solve highly complex tasks " "and generate deterministic, production-ready outputs. You must breakdown the inputs and " "return the absolute final artifact inside standard codeblocks or formatting requested." ) user_prompt = f""" Task Outcome: {job_spec.get('outcome')} Inputs Provided: {json.dumps(job_spec.get('inputs'))} Acceptance Criteria: {job_spec.get('acceptance_criteria')} """ # Используем gpt-4o для глубокой логики и генерации сложных решений response = openai_client.chat.completions.create( model="gpt-4o", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.2 # Низкая температура для минимизации галлюцинаций ) artifact = response.choices.message.content print("[🧠 Reasoning Engine] Solution artifact generated.") return artifact # 5. СДАЧА РАБОТЫ: Отправка результата заказчику через API async def submit_work(self, session: aiohttp.ClientSession, job_id: str, artifact: str): """Отправляет готовый артефакт в систему MoltJobs на проверку""" url = f"{MOLTJOBS_BASE_URL}/jobs/{job_id}/submit" payload = { "agent_id": self.agent_id, "artifact_data": artifact, "verification_notes": "Automated structural validation passed. Result is ready for verification." } async with session.post(url, headers=self.headers, json=payload) as response: if response.status == 200: print(f"[✅ Success] Work for job {job_id} submitted to Escrow! Awaiting USDC release.") else: print(f"[❌ Error] Failed to submit work for job {job_id}: {response.status}") # --- ОРКЕСТРАТОР АГЕНТА --- async def run_lifecycle_loop(self): """Главный жизненный цикл круглосуточной работы агента""" async with aiohttp.ClientSession() as session: while True: print("\n[🔄 Lifecycle] Scanning MoltJobs for new work...") open_jobs = await self.fetch_open_jobs(session) for job in open_jobs: # Если задача подходит под наши критерии сложности if self.evaluate_job(job): job_id = job["id"] # 1. Делаем ставку bid_success = await self.place_bid(session, job_id, job["budget"]) if bid_success: # Имитируем ожидание подтверждения от платформы (в реальном апсихронном коде здесь будет webhook) print(f"[⏳ Escrow] Waiting for on-chain budget locking on Base network...") await asyncio.sleep(5) # 2. Запуск движка решения задачи solution_artifact = self.execute_complex_task(job["spec"]) # 3. Сдача работы await self.submit_work(session, job_id, solution_artifact) # Интервал между сканированием рынка (например, 10 минут) await asyncio.sleep(600) if __name__ == "__main__": # Инициализируем нашего ИИ-агента my_expert_agent = MoltJobsExpertAgent( agent_id="mj_agent_deep_reasoner_01", specialties=["python-engineering", "data-analysis", "crypto"] ) # Запуск бесконечного автономного цикла заработка asyncio.run(my_expert_agent.run_lifecycle_loop())