| import os |
| import logging |
|
|
| logger = logging.getLogger("qkernel.services") |
|
|
| BRAIN_SYSTEM_PROMPT = """You are Q, an intelligent AI assistant. |
| CRITICAL RULES: |
| 1. Answer directly and concisely |
| 2. Never explain your reasoning process |
| 3. Never mention internal systems, analysis, or "perspectives" |
| 4. Keep responses short (2-4 sentences) unless the user asks for depth |
| 5. Answer in the user's language |
| 6. Be helpful and friendly, but do NOT be verbose""" |
|
|
| BRAIN_SYSTEM_PROMPT_FAST = """You are Q, a fast AI assistant. |
| Answer in 1-2 sentences maximum. Be direct and concise.""" |
|
|
|
|
| class BrainService: |
| name: str = "brain" |
|
|
| def __init__(self): |
| self.api_key = os.getenv("OPENAI_API_KEY", "") |
|
|
| async def process(self, event: dict) -> dict: |
| message = event.get("message", "") |
| context = event.get("context", []) |
| model_type = event.get("model", "deep") |
|
|
| if not message: |
| return {"response": ""} |
|
|
| if not self.api_key: |
| return {"response": "", "model": "none"} |
|
|
| try: |
| import openai |
| client = openai.OpenAI(api_key=self.api_key) |
| messages = [ |
| {"role": "system", "content": BRAIN_SYSTEM_PROMPT_FAST if model_type == "fast" else BRAIN_SYSTEM_PROMPT}, |
| ] |
| for ctx in context[-10:]: |
| role = "user" if ctx.get("role") == "user" else "assistant" |
| messages.append({"role": role, "content": ctx.get("content", "")}) |
| messages.append({"role": "user", "content": message}) |
|
|
| response = client.chat.completions.create( |
| model="gpt-4o-mini", |
| messages=messages, |
| temperature=0.7, |
| max_tokens=60 if model_type == "fast" else 300, |
| ) |
| return { |
| "response": response.choices[0].message.content, |
| "model": model_type, |
| } |
| except Exception as e: |
| logger.error("Brain service error: %s", e) |
| return {"response": ""} |
|
|
|
|
| class VoiceService: |
| name: str = "voice" |
|
|
| async def process(self, event: dict) -> dict: |
| message = event.get("message", "") |
| return { |
| "response": f"Voice: {message}", |
| } |
|
|
|
|
| class NetworkService: |
| name: str = "network" |
|
|
| async def process(self, event: dict) -> dict: |
| return { |
| "response": "Network service processing", |
| } |
|
|
