File size: 2,438 Bytes
945aae0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | 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",
}
|