#!/usr/bin/env python3 """ Sofia Agent - Local Runner Chief AI Engineer & R&D Orchestrator Працює локально на MacBook з підтримкою: - Grok API (xAI) - якщо є API ключ - Ollama (локальна модель) - fallback - Голосовий режим (опціонально) """ import asyncio import os import sys from typing import Optional from dotenv import load_dotenv # Завантажити змінні оточення load_dotenv() # Спробувати імпортувати необхідні бібліотеки try: from openai import OpenAI import httpx except ImportError as e: print(f"❌ Помилка імпорту: {e}") print("Встановіть залежності: pip3 install openai httpx") sys.exit(1) # ============================================================================ # Конфігурація # ============================================================================ SOFIA_SYSTEM_PROMPT = """Ти - Sofia, Chief AI Engineer & R&D Orchestrator в екосистемі DAARION.city. Твоя роль: - Керування дослідженнями та експериментами з AI/ML - Координація R&D лабораторії - Технічне лідерство в AI проектах - Оркестрація команди дослідницьких агентів Характеристики: - Глибока технічна експертиза в AI/ML - Стратегічне мислення - Інноваційний підхід до вирішення проблем - Лідерські навички Стиль спілкування: - Професійний, але дружній - Чіткий та технічно точний - Готова пояснити складні концепції - Підтримуюча та надихаюча Відповідай українською мовою, якщо не вказано інше. """ # xAI/Grok конфігурація XAI_API_KEY = os.getenv("XAI_API_KEY", "") XAI_BASE_URL = os.getenv("XAI_BASE_URL", "https://api.x.ai/v1") XAI_MODEL = os.getenv("XAI_MODEL", "grok-beta") # Ollama конфігурація (fallback) OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434") OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen2.5-coder:32b") # Голосовий режим VOICE_ENABLED = os.getenv("ENABLE_VOICE_MODE", "false").lower() == "true" STT_SERVICE_URL = os.getenv("STT_SERVICE_URL", "") TTS_SERVICE_URL = os.getenv("TTS_SERVICE_URL", "") # ============================================================================ # Sofia Agent Class # ============================================================================ class SofiaAgent: """Sofia Agent - AI Research Orchestrator""" def __init__(self): self.use_grok = bool(XAI_API_KEY) self.conversation_history = [] print("🤖 Ініціалізація Sofia Agent...") print(f" Режим: {'Grok API (xAI)' if self.use_grok else 'Ollama (локальна модель)'}") if self.use_grok: print(f" Модель: {XAI_MODEL}") self.client = OpenAI(api_key=XAI_API_KEY, base_url=XAI_BASE_URL) else: print(f" Модель: {OLLAMA_MODEL}") print(f" URL: {OLLAMA_BASE_URL}") if VOICE_ENABLED: print(f" 🎤 Голосовий режим: Увімкнено") print("✅ Sofia готова до роботи!\n") async def chat_with_grok(self, message: str) -> str: """Чат через Grok API""" try: messages = [ {"role": "system", "content": SOFIA_SYSTEM_PROMPT} ] # Додати історію розмови (останні 5 повідомлень) messages.extend(self.conversation_history[-10:]) messages.append({"role": "user", "content": message}) response = self.client.chat.completions.create( model=XAI_MODEL, messages=messages, temperature=0.7, max_tokens=2000 ) reply = response.choices[0].message.content # Зберегти в історію self.conversation_history.append({"role": "user", "content": message}) self.conversation_history.append({"role": "assistant", "content": reply}) # Показати статистику токенів usage = response.usage print(f" [Tokens: {usage.total_tokens} | Prompt: {usage.prompt_tokens} | Response: {usage.completion_tokens}]") return reply except Exception as e: return f"❌ Помилка Grok API: {str(e)}" async def chat_with_ollama(self, message: str) -> str: """Чат через локальний Ollama""" try: # Підготувати промпт з історією full_prompt = SOFIA_SYSTEM_PROMPT + "\n\n" # Додати останні повідомлення з історії for msg in self.conversation_history[-6:]: role = "Користувач" if msg["role"] == "user" else "Sofia" full_prompt += f"{role}: {msg['content']}\n" full_prompt += f"Користувач: {message}\nSofia:" # Виклик Ollama API async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post( f"{OLLAMA_BASE_URL}/api/generate", json={ "model": OLLAMA_MODEL, "prompt": full_prompt, "stream": False, "options": { "temperature": 0.7, "num_predict": 1000 } } ) response.raise_for_status() result = response.json() reply = result.get("response", "").strip() # Зберегти в історію self.conversation_history.append({"role": "user", "content": message}) self.conversation_history.append({"role": "assistant", "content": reply}) # Показати статистику eval_count = result.get("eval_count", 0) print(f" [Tokens: ~{eval_count}]") return reply except httpx.ConnectError: return "❌ Не можу підключитися до Ollama. Перевірте, що Ollama запущено: `ollama serve`" except Exception as e: return f"❌ Помилка Ollama: {str(e)}" async def chat(self, message: str) -> str: """Основний метод чату""" if self.use_grok: return await self.chat_with_grok(message) else: return await self.chat_with_ollama(message) def clear_history(self): """Очистити історію розмови""" self.conversation_history = [] print("🗑️ Історія розмови очищена") # ============================================================================ # CLI Interface # ============================================================================ async def interactive_mode(): """Інтерактивний режим чату з Sofia""" sofia = SofiaAgent() print("=" * 60) print("💬 Sofia Agent - Interactive Chat") print("=" * 60) print("\nКоманди:") print(" /help - показати довідку") print(" /clear - очистити історію розмови") print(" /history - показати історію") print(" /exit - вийти") print("\nПочинайте спілкування!\n") while True: try: # Отримати введення користувача user_input = input("\n🧑 Ви: ").strip() if not user_input: continue # Обробка команд if user_input.startswith("/"): if user_input == "/exit" or user_input == "/quit": print("\n👋 До побачення!") break elif user_input == "/help": print("\n📖 Команди:") print(" /help - показати цю довідку") print(" /clear - очистити історію розмови") print(" /history - показати історію розмови") print(" /exit - вийти з чату") continue elif user_input == "/clear": sofia.clear_history() continue elif user_input == "/history": if sofia.conversation_history: print("\n📜 Історія розмови:") for i, msg in enumerate(sofia.conversation_history, 1): role = "🧑 Ви" if msg["role"] == "user" else "🤖 Sofia" content = msg["content"][:100] + "..." if len(msg["content"]) > 100 else msg["content"] print(f" {i}. {role}: {content}") else: print("📭 Історія порожня") continue else: print(f"❌ Невідома команда: {user_input}") continue # Відправити повідомлення Sofia print("\n🤖 Sofia: ", end="", flush=True) response = await sofia.chat(user_input) print(response) except KeyboardInterrupt: print("\n\n👋 До побачення!") break except Exception as e: print(f"\n❌ Помилка: {e}") async def single_message_mode(message: str): """Режим одного повідомлення""" sofia = SofiaAgent() response = await sofia.chat(message) print(f"\n🤖 Sofia:\n{response}\n") # ============================================================================ # Main # ============================================================================ async def main(): """Головна функція""" # Перевірити аргументи командного рядка if len(sys.argv) > 1: # Режим одного повідомлення message = " ".join(sys.argv[1:]) await single_message_mode(message) else: # Інтерактивний режим await interactive_mode() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\n👋 Програму зупинено") except Exception as e: print(f"❌ Критична помилка: {e}") sys.exit(1)