Backup code/summarize_memory_21.py

# -*- coding: utf-8 -*-

import sys
import os
import json
import time
import copy
import argparse
from typing import Dict, Any
from openai import OpenAI
import re
import openai

# Add memory bank path
sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'memory_bank'))

# Configuration for OpenAI API
CHATGPT_CONFIG = {
    "model": "gpt-4o",
    "temperature": 0.7,
    "max_tokens": 400,
    "top_p": 1.0,
    "frequency_penalty": 0.4,
    "presence_penalty": 0.2,
    "stop": ["<|im_end|>", "¬Human"]
}

# Default memory directory
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
MEMORY_DIR = os.path.join(BASE_DIR, "memories", "update_memory_0512_eng.json")

class LLMClientSimple:
    def __init__(self, config, gen_config=None):
        self.config = config
        self.client = OpenAI(
            api_key=self.config.get("api_key") or os.environ.get("GAPGPT_API_KEY"),
            base_url=self.config.get("base_url") or os.environ.get(
                "GAPGPT_BASE_URL", "https://api.gapgpt.app/v1"
            ),
        )
        self.disable_tqdm = False
        self.gen_config = copy.deepcopy(
            gen_config if gen_config is not None else config
        )

        self.disable_tqdm = False
        self.gen_config = gen_config

    def generate_text_simple(self, prompt, prompt_num, language='en'):
        self.gen_config['n'] = prompt_num
        retry_times, count = 5, 0
        response = None
        
        while response is None and count < retry_times:
            try:
                request = copy.deepcopy(self.gen_config)
                
                # System prompt setup based on language
                if language == 'cn':
                    message = [
                        {"role": "system", "content": "The following is a transcript of a conversation between a human and a smart, psychology-savvy AI assistant."},
                        {"role": "user", "content": "Hello! Please help me summarize the conversation"},
                        {"role": "system", "content": "OK, I'll try to help you."},
                        {"role": "user", "content": f"{prompt}"}
                    ]
                else:
                    message = [
                        {"role": "system", "content": "Below is a transcript of a conversation between a human and an AI assistant that is intelligent and knowledgeable in psychology."},
                        {"role": "user", "content": "Hello! Please help me summarize the content of the conversation."},
                        {"role": "system", "content": "Sure, I will do my best to assist you."},
                        {"role": "user", "content": f"{prompt}"}
                    ]

                # Call the API method
                response = self.client.chat.completions.create(**request, messages=message)

            except Exception as e:
                print(e)
                # Check for context limit errors
                if 'context' in str(e).lower() or 'maximum' in str(e).lower():
                    cut_length = 1800 - 200 * count
                    print('Max context length reached, cut to {}'.format(cut_length))
                    prompt = prompt[-cut_length:]
                    response = None
                count += 1
        
        if response:
            # Get response content using dot notation
            task_desc = response.choices[0].message.content
        else:
            task_desc = ''
            
        return task_desc


# Lazy initialization instead of module-level
_llm_client = None

def get_llm_client():
    global _llm_client
    if _llm_client is None:
        _llm_client = LLMClientSimple(CHATGPT_CONFIG, gen_config=CHATGPT_CONFIG)
    return _llm_client



# --- Prompt Generation Functions ---

def summarize_content_prompt(content, user_name, boot_name, language='en'):
    # Base prompt adjusted based on psychology
    prompt = 'Please summarize the following dialogue from a psychological perspective, focusing on the user\'s emotional state, key concerns, and potential therapeutic insights. Dialogue content:\n'
    
    for dialog in content:
        query = dialog['query']
        response = dialog['response']
        prompt += f"\n{user_name}: {query.strip()}"
        prompt += f"\n{boot_name}: {response.strip()}"
    
    prompt += ('\nSummary:' if language == 'cn' else '\nSummarization:')
    return prompt


def summarize_user_issues_prompt(content, user_name, boot_name, language='en'):
    """
    Analyze the conversation from a psychological perspective, identify user problems and their causes,
    and organize them into a structured JSON format.
    """
    prompt = (
        "Please analyze the following dialogue from a psychological perspective. Identify the user's key "
        "problems and the underlying causes behind them. Structure the results in a clear and concise way. "
        "Dialogue content:\n"
    )

    for dialog in content:
        query = dialog['query']
        response = dialog['response']
        prompt += f"\n{user_name}: {query.strip()}"
        prompt += f"\n{boot_name}: {response.strip()}"

    prompt += '\n\nPlease list the user’s problems and their causes in JSON format, like this:\n'
    prompt += '{\n  "problems": [\n    {\n      "problem": "<problem_1>",\n      "cause": "<cause_1>"\n    },\n    {\n      "problem": "<problem_2>",\n      "cause": "<cause_2>"\n    }\n  ]\n}'

    return prompt


def summarize_overall_prompt(content, language='en'):
    prompt = 'Please summarize the following events in a high-level manner, and try to be as concise as possible, summarizing and retaining the core key information:\n' if language == 'cn' else "Please provide a highly concise summary of the following event, capturing the essential key information as succinctly as possible. Summarize the event:\n"
    
    for date, summary_dict in content:
        summary = summary_dict['content']
        prompt += (f"\nTime {date} The event that occurred was {summary.strip()}" if language == 'cn' else f"At {date}, the events are {summary.strip()}")
    
    prompt += ('\nSummary:' if language == 'cn' else '\nSummarization:')
    return prompt


def summarize_overall_personality(content, language='en'):
    prompt = 'Below are the personality traits and moods of users in multiple conversations, as well as the appropriate response strategies at the time:\n' if language == 'cn' else "The following are the user's exhibited personality traits and emotions throughout multiple dialogues, along with appropriate response strategies for the current situation:"
    
    for date, summary in content:
        prompt += (f"\n in time {date} The analysis is {summary.strip()}" if language == 'cn' else f"At {date}, the analysis shows {summary.strip()}")
    
    prompt += ('\nPlease give a general summary of the users personality and the most appropriate response strategy for the AI lover, and try to be concise and highly summarized. The summary is:' if language == 'cn' else "Please provide a highly concise and general summary of the user's personality and the most appropriate response strategy for the AI lover, summarized as:")
    return prompt


def summarize_person_prompt(content, user_name, boot_name, language):
    prompt = f"Based on the following dialogue, analyze {user_name}'s emotional state, cognitive patterns, and potential psychological needs. Suggest therapeutic strategies for the AI to respond empathetically and effectively. Dialogue content:\n"
    
    for dialog in content:
        query = dialog['query']
        response = dialog['response']
        prompt += f"\n{user_name}: {query.strip()}"
        prompt += f"\n{boot_name}: {response.strip()}"

    prompt += (f'\n{user_name} personality traits, mood, {boot_name} The response strategy is:' if language == 'cn' else f"\n{user_name}'s personality traits, emotions, and {boot_name}'s response strategy are:")
    return prompt


# --- Memory Extraction Functions ---

def extract_session_summary(conversation_text, session_date, session_id):
    """
    Extracts session-level insights and creates an episodic memory entry.

    Args:
        conversation_text (str): The full conversation from the session.
        session_date (str): The date of the session.
        session_id (int): The index of the session.

    Returns:
        dict: Extracted episodic memory entry.
    """
    # Prompt to the LLM to extract insights
    insight_prompt = f"""
Please analyze the following conversation and extract key information:

Conversation:
{conversation_text}

1. Topics discussed (as a list)
2. User's emotional state (e.g., happy, anxious, stressed)
3. Key insights or takeaways (what the user is feeling or thinking)

Format response as JSON:
{{
  "topics_discussed": [],
  "emotional_state": "",
  "insights": ""
}}
"""

    response = get_llm_client().generate_text_simple(
        prompt=insight_prompt,
        prompt_num=1,
        language="en",
    )

    # Remove extra whitespace
    response = response.strip()

    # حذف تگ شروع بلاک کد مارک‌داون
    if response.startswith("```json"):response = response[7:]
    elif response.startswith("```"):response = response[3:]

    # حذف تگ پایان بلاک کد مارک‌داون
    if response.endswith("```"):
        response = response[:-3]

        response = response.strip()

    try:
        episodic_memory = json.loads(response)
    except json.JSONDecodeError:
        print("⚠️ Failed to parse episodic memory response!")
        episodic_memory = {
        "topics_discussed": ["N/A"],
        "emotional_state": "unknown",
        "insights": "No insights extracted.",
        }

    # Add session metadata
    episodic_memory["session_date"] = session_date
    episodic_memory["session_id"] = session_id

    return episodic_memory



def extract_semantic_memory(latest_episodic_memory, existing_semantic_memory):
    """
    Updates semantic memory based on the latest episodic memory.
    """
    # Use the insights from episodic memory to infer stable traits and patterns
    semantic_prompt = f"""
    Please analyze the following session summary and infer long-term personality traits and stable characteristics:
    Session Summary:
    {json.dumps(latest_episodic_memory, indent=4)}

    Update the user's semantic memory with:
    1. Evolving personality traits (Big Five).
    2. Core values and recurring motivations.
    3. Behavioral patterns and consistent emotional responses.
    4. Frequent topics and recurring themes.

    Format response as JSON:
    {{
    "personality_traits": {{
    "openness": 0.0,
    "conscientiousness": 0.0,
    "extraversion": 0.0,
    "agreeableness": 0.0,
    "neuroticism": 0.0
    }},
    "core_values": [],
    "behavioral_patterns": [],
    "recurring_themes": []
    }}
    """

    response = get_llm_client().generate_text_simple(prompt=semantic_prompt, prompt_num=1, language='en')

    # Safe JSON extraction using Regex
    try:
        # Find content between first { and last }
        json_match = re.search(r'\{.*\}', response, re.DOTALL)
        if json_match:
            json_str = json_match.group(0)
            new_semantic_data = json.loads(json_str)
        else:
            raise ValueError("No JSON object found in response.")

    except Exception as e:
        print(f"⚠️ Failed to parse semantic memory response! Error: {e}")
        print(f"LLM Response was:\n{response}") 
        return existing_semantic_memory

    # Merge new data into semantic memory
    updated_semantic_memory = existing_semantic_memory if isinstance(existing_semantic_memory, dict) else {}

    # Update or average personality traits
    for trait, value in new_semantic_data.get("personality_traits", {}).items():
        try:
            val_float = float(value)
        except (ValueError, TypeError):
            val_float = 0.0

        if trait in updated_semantic_memory.get("personality_traits", {}):
            existing_value = float(updated_semantic_memory["personality_traits"].get(trait, 0.0))
            updated_semantic_memory["personality_traits"][trait] = round((existing_value + val_float) / 2, 2)
        else:
            updated_semantic_memory.setdefault("personality_traits", {})[trait] = val_float

    # Combine core values, avoiding duplicates
    updated_semantic_memory["core_values"] = list(set(
        updated_semantic_memory.get("core_values", []) + new_semantic_data.get("core_values", [])
    ))

    # Combine behavioral patterns
    updated_semantic_memory["behavioral_patterns"] = list(set(
        updated_semantic_memory.get("behavioral_patterns", []) + new_semantic_data.get("behavioral_patterns", [])
    ))

    # Combine recurring themes
    updated_semantic_memory["recurring_themes"] = list(set(
        updated_semantic_memory.get("recurring_themes", []) + new_semantic_data.get("recurring_themes", [])
    ))

    print("\n✅ Semantic memory updated successfully!")
    return updated_semantic_memory

    # Merge new data into semantic memory
    updated_semantic_memory = existing_semantic_memory if isinstance(existing_semantic_memory, dict) else {}

    # Update or average personality traits
    for trait, value in new_semantic_data.get("personality_traits", {}).items():
        try:
            val_float = float(value)
        except (ValueError, TypeError):
            val_float = 0.0

        if trait in updated_semantic_memory.get("personality_traits", {}):
            existing_value = float(updated_semantic_memory["personality_traits"].get(trait, 0.0))
            updated_semantic_memory["personality_traits"][trait] = round((existing_value + val_float) / 2, 2)
        else:
            updated_semantic_memory.setdefault("personality_traits", {})[trait] = val_float

    # Combine core values, avoiding duplicates
    updated_semantic_memory["core_values"] = list(set(
        updated_semantic_memory.get("core_values", []) + new_semantic_data.get("core_values", [])
    ))

    # Combine behavioral patterns
    updated_semantic_memory["behavioral_patterns"] = list(set(
        updated_semantic_memory.get("behavioral_patterns", []) + new_semantic_data.get("behavioral_patterns", [])
    ))

    # Combine recurring themes
    updated_semantic_memory["recurring_themes"] = list(set(
        updated_semantic_memory.get("recurring_themes", []) + new_semantic_data.get("recurring_themes", [])
    ))

    print("\n✅ Semantic memory updated successfully!")
    return updated_semantic_memory


# --- Main Logic ---

def summarize_memory(memory_dir, name=None, language='en'):
    boot_name = 'AI'
    gen_prompt_num = 1

    with open(memory_dir, 'r', encoding='utf8') as f:
        memory = json.loads(f.read())

    for k, v in memory.items():
        if name is not None and k != name:
            continue

        user_name = k
        print(f'Updating memory for user {user_name}')

        if v.get('history') is None:
            continue

        history = v['history']

        if v.get('summary') is None:
            memory[user_name]['summary'] = {}
        if v.get('personality') is None:
            memory[user_name]['personality'] = {}
        if v.get('issues') is None:
            memory[user_name]['issues'] = {}

        for date, content in history.items():
            # Check flags to see if update is needed
            his_flag = False if (date in v['summary'].keys() and v['summary'][date]) else True
            person_flag = False if (date in v['personality'].keys() and v['personality'][date]) else True
            problem_flag = False if (date in v['issues'].keys() and v['issues'][date]) else True 

            hisprompt = summarize_content_prompt(content, user_name, boot_name, language)
            issues_prompt = summarize_user_issues_prompt(content, user_name, boot_name, language)
            person_prompt = summarize_person_prompt(content, user_name, boot_name, language)

            if his_flag:
                his_summary = get_llm_client().generate_text_simple(prompt=hisprompt, prompt_num=gen_prompt_num, language=language)
                memory[user_name]['summary'][date] = {'content': his_summary}

            if problem_flag:
                issues_summary = get_llm_client().generate_text_simple(prompt=issues_prompt, prompt_num=gen_prompt_num, language=language)
                memory[user_name]['issues'][date] = {'content': issues_summary}
                print(issues_summary)

            if person_flag:
                person_summary = get_llm_client().generate_text_simple(prompt=person_prompt, prompt_num=gen_prompt_num, language=language)
                memory[user_name]['personality'][date] = person_summary

        overall_his_prompt = summarize_overall_prompt(list(memory[user_name]['summary'].items()), language=language)
        overall_person_prompt = summarize_overall_personality(list(memory[user_name]['personality'].items()), language=language)

        memory[user_name]['overall_history'] = get_llm_client().generate_text_simple(prompt=overall_his_prompt, prompt_num=gen_prompt_num, language=language)
        memory[user_name]['overall_personality'] = get_llm_client().generate_text_simple(prompt=overall_person_prompt, prompt_num=gen_prompt_num, language=language)
     
    with open(memory_dir, 'w', encoding='utf8') as f:
        print(f'Successfully updated memory for {name}')
        json.dump(memory, f, ensure_ascii=False)
        
    return memory


if __name__ == '__main__':
    target_memory = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'memories', 'eng_memory_cases.json')
    summarize_memory(target_memory, language='en')