"""
This module provides functionality for interacting with the Claude AI API.
It includes functions for loading configurations, constructing prompts,
sending requests to the Claude API, and managing message history.
"""
import aiohttp
import json
import os
import yaml
import datetime
from collections import deque
from typing import Dict, List
# Constants
MAX_CONTEXT_MESSAGES = 10
ANTHROPIC_VERSION = '2023-06-01'
CONFIG_FILE = 'config.yml'
PERSONAS_FILE = 'twitch_claude_bot/personas/claude_personas.json'
LOG_FILE = 'claude_interactions.log'
# Load configuration
[docs]
def load_config(file_name: str) -> Dict:
"""
Load configuration from a YAML file.
Args:
file_name (str): The name of the configuration file.
Returns:
Dict: A dictionary containing the configuration data.
"""
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
file_path = os.path.join(script_dir, file_name)
try:
with open(file_path, 'r') as file:
return yaml.safe_load(file)
except FileNotFoundError:
print(f"Error: '{file_name}' not found at {file_path}")
return {}
config = load_config(CONFIG_FILE)
CLAUDE_API_KEY = config.get('ai_engines', {}).get('claude', {}).get('api_key', '')
CLAUDE_API_URL = 'https://api.anthropic.com/v1/messages'
# Load personas
CLAUDE_PERSONAS = load_config(PERSONAS_FILE)
message_history = deque(maxlen=MAX_CONTEXT_MESSAGES)
[docs]
async def get_claude_response(prompt: str, persona: str = 'claude_the_wizard') -> str:
"""
Get a response from the Claude AI based on the given prompt and persona.
Args:
prompt (str): The user's input prompt.
persona (str, optional): The persona to use for the response. Defaults to 'claude_the_wizard'.
Returns:
str: The AI-generated response.
Raises:
Exception: If the API request fails.
"""
persona_config = CLAUDE_PERSONAS.get(persona, {})
full_prompt = construct_full_prompt(prompt, persona_config)
conversation = construct_conversation(full_prompt)
async with aiohttp.ClientSession() as session:
headers = {
'Content-Type': 'application/json',
'x-api-key': CLAUDE_API_KEY,
'anthropic-version': ANTHROPIC_VERSION
}
data = {
'model': 'claude-2.1',
'max_tokens': 500,
'messages': conversation
}
async with session.post(CLAUDE_API_URL, headers=headers, json=data) as response:
if response.status == 200:
result = await response.json()
response_text = result['content'][0]['text']
update_message_history(prompt, response_text)
log_interaction(prompt, response_text)
return response_text
else:
error_message = await response.text()
raise Exception(f"API request failed with status {response.status}: {error_message}")
[docs]
def construct_full_prompt(prompt: str, persona_config: Dict) -> str:
"""
Construct a full prompt by combining the user's input with persona configuration.
Args:
prompt (str): The user's input prompt.
persona_config (Dict): The configuration for the selected persona.
Returns:
str: The full constructed prompt.
"""
prompt_parts = [
persona_config.get('prompt_prefix', ''),
f"Name: {persona_config.get('name', '')}",
f"Description: {persona_config.get('description', '')}",
f"Background: {persona_config.get('background', '')}",
f"Personality traits: {', '.join(persona_config.get('personality_traits', []))}",
f"Knowledge areas: {', '.join(persona_config.get('knowledge_areas', []))}",
f"Speech patterns: {', '.join(persona_config.get('speech_patterns', []))}",
prompt,
persona_config.get('prompt_suffix', '')
]
return '\n\n'.join(prompt_parts)
[docs]
def construct_conversation(full_prompt: str) -> List[Dict]:
"""
Construct a conversation history including the new prompt.
Args:
full_prompt (str): The full constructed prompt.
Returns:
List[Dict]: A list of conversation messages in the format required by the API.
"""
conversation = [
{"role": 'user' if i % 2 == 0 else 'assistant', "content": msg}
for i, msg in enumerate(message_history)
]
conversation.append({"role": "user", "content": full_prompt})
return conversation
[docs]
def update_message_history(prompt: str, response: str) -> None:
"""
Update the message history with the new prompt and response.
Args:
prompt (str): The user's input prompt.
response (str): The AI-generated response.
"""
message_history.extend([prompt, response])
[docs]
def log_interaction(prompt: str, response: str) -> None:
"""
Log the interaction between the user and the AI.
Args:
prompt (str): The user's input prompt.
response (str): The AI-generated response.
"""
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}]\nPrompt: {prompt}\nFull Answer: {response}\n\n"
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
log_file_path = os.path.join(script_dir, LOG_FILE)
with open(log_file_path, 'a', encoding='utf-8') as log_file:
log_file.write(log_entry)
[docs]
def clear_message_history() -> None:
"""Clear the message history."""
message_history.clear()