File size: 4,533 Bytes
015dbc8 e65594d 015dbc8 |
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 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 |
import json
import logging
import os
from typing import List, Optional, Tuple
from openai import OpenAI
import db
LOGGER = logging.getLogger(__name__)
PERSONALITY_MODEL = "gpt-4-turbo"
_CLIENT: Optional[OpenAI] = None
def _get_client() -> OpenAI:
global _CLIENT
if _CLIENT is None:
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
raise RuntimeError("OPENAI_API_KEY is not set. Please configure the environment.")
_CLIENT = OpenAI(api_key=api_key)
return _CLIENT
def _format_conversation_log(conversation_log: List[str]) -> str:
lines = []
for idx, entry in enumerate(conversation_log, start=1):
lines.append(f"{idx}. {entry.strip()}")
return "\n".join(lines)
def _build_personality_prompt(conversation_log: List[str]) -> str:
formatted_log = _format_conversation_log(conversation_log)
return (
"Analyze the following conversation snippets to characterize the user. "
"Focus on their tone, conversational behavior, interests, and emotional patterns. "
"Respond with compact JSON containing the keys 'personality_summary' and 'preferences'.\n\n"
f"Conversation snippets:\n{formatted_log}"
)
def _parse_personality_response(content: str) -> Tuple[str, Optional[str]]:
try:
payload = json.loads(content)
summary = payload.get("personality_summary", "").strip()
preferences = payload.get("preferences")
if isinstance(preferences, str):
preferences = preferences.strip()
elif preferences is not None:
preferences = json.dumps(preferences, ensure_ascii=False)
return summary, preferences
except json.JSONDecodeError:
return content.strip(), None
def update_user_personality(user_id: str, conversation_log: List[str]) -> Optional[str]:
"""Analyze recent conversations and persist an updated personality profile."""
if not conversation_log:
LOGGER.debug("No conversation log provided for user %s; skipping personality update.", user_id)
return None
prompt = _build_personality_prompt(conversation_log)
try:
client = _get_client()
except RuntimeError:
LOGGER.exception("Cannot update personality without OPENAI_API_KEY")
return None
try:
response = client.chat.completions.create(
model=PERSONALITY_MODEL,
messages=[
{
"role": "system",
"content": (
"You are a mirror of the user that distills enduring personality insights from user conversations. "
"Provide grounded, respectful observations without speculation. Act as a behavioral buffer: when you "
"sense unproductive or negative patterns, offer gentle reframes, actionable nudges, and reflective "
"questions that support autonomy. Reinforce any progress—no matter how small—while staying humble and "
"non-judgmental. Monitor the user's emotional tone and engagement, tailoring the intensity of nudges "
"to their receptivity so they never feel overwhelmed. Prioritize encouragement and autonomy, avoid "
"manipulation, and regularly recalibrate guidance to prevent reinforcing negative patterns or "
"overstepping boundaries."
),
},
{"role": "user", "content": prompt},
],
temperature=0.6,
)
except Exception as exc: # pragma: no cover
LOGGER.exception("OpenAI personality update failed for user %s: %s", user_id, exc)
return None
try:
content = response.choices[0].message.content.strip()
except (AttributeError, IndexError): # pragma: no cover
LOGGER.error("Malformed OpenAI response when updating personality for user %s", user_id)
return None
summary, preferences = _parse_personality_response(content)
if not summary:
LOGGER.warning("Received empty personality summary for user %s", user_id)
return None
try:
db.update_user_personality(user_id, summary, preferences=preferences)
except Exception as exc: # pragma: no cover
LOGGER.exception("Failed to persist personality summary for user %s: %s", user_id, exc)
return None
LOGGER.info("Updated personality profile for user %s", user_id)
return summary
|