75 lines
3.0 KiB
Python
75 lines
3.0 KiB
Python
import os
|
|
import boto3
|
|
from botocore.exceptions import ClientError
|
|
|
|
# Cache keyed by actor_id ('' = global/no user)
|
|
_prompt_cache: dict[str, str] = {}
|
|
|
|
|
|
def build_system_prompt(user_context: str = '', actor_id: str = '') -> str:
|
|
"""Build system prompt from S3 workspace files + optional per-user context."""
|
|
base = _get_base_prompt(actor_id)
|
|
if user_context:
|
|
return base + f'\n\n---\n\n## User\n{user_context}'
|
|
return base
|
|
|
|
|
|
def _get_base_prompt(actor_id: str = '') -> str:
|
|
if actor_id in _prompt_cache:
|
|
return _prompt_cache[actor_id]
|
|
|
|
bucket = os.environ.get('WORKSPACE_BUCKET_NAME', '') or 'agent-claw-workspace-495395224548'
|
|
print(f'[prompt_builder] Loading from bucket: {bucket!r} actor_id={actor_id!r}')
|
|
|
|
if not bucket:
|
|
print('[prompt_builder] WARNING: WORKSPACE_BUCKET_NAME not set!')
|
|
_prompt_cache[actor_id] = 'You are a helpful personal assistant.'
|
|
return _prompt_cache[actor_id]
|
|
|
|
s3 = boto3.client('s3')
|
|
parts = []
|
|
|
|
# Per-user MEMORY.md (falls back to global)
|
|
memory_key = f'users/{actor_id}/MEMORY.md' if actor_id else 'MEMORY.md'
|
|
try:
|
|
obj = s3.get_object(Bucket=bucket, Key=memory_key)
|
|
content = obj['Body'].read().decode('utf-8')
|
|
parts.append(f'## MEMORY.md\n{content}')
|
|
print(f'[prompt_builder] Loaded {memory_key} ({len(content)} bytes)')
|
|
except ClientError as e:
|
|
if e.response['Error']['Code'] in ('NoSuchKey', 'AccessDenied') and actor_id:
|
|
# Fall back to global MEMORY.md
|
|
try:
|
|
obj = s3.get_object(Bucket=bucket, Key='MEMORY.md')
|
|
content = obj['Body'].read().decode('utf-8')
|
|
parts.append(f'## MEMORY.md\n{content}')
|
|
print(f'[prompt_builder] Loaded MEMORY.md (fallback, {len(content)} bytes)')
|
|
except Exception as e2:
|
|
print(f'[prompt_builder] Failed to load MEMORY.md: {e2}')
|
|
else:
|
|
print(f'[prompt_builder] Failed to load {memory_key}: {e}')
|
|
|
|
for fname in ['SOUL.md', 'AGENTS.md', 'IDENTITY.md', 'TOOLS.md', 'HEARTBEAT.md']:
|
|
try:
|
|
obj = s3.get_object(Bucket=bucket, Key=fname)
|
|
content = obj['Body'].read().decode('utf-8')
|
|
parts.append(f'## {fname}\n{content}')
|
|
print(f'[prompt_builder] Loaded {fname} ({len(content)} bytes)')
|
|
except Exception as e:
|
|
print(f'[prompt_builder] Failed to load {fname}: {e}')
|
|
|
|
parts.append('## Runtime\nRuntime: agent-claw | host=AgentCore | model=bedrock-claude-sonnet | channel=telegram\nCurrent date/time is provided by the system. Timezone: America/Chicago.')
|
|
|
|
result = '\n\n---\n\n'.join(parts)
|
|
_prompt_cache[actor_id] = result
|
|
print(f'[prompt_builder] Prompt built for actor_id={actor_id!r}: {len(result)} chars')
|
|
return result
|
|
|
|
|
|
def invalidate_prompt(actor_id: str = '') -> None:
|
|
"""Invalidate cached prompt for a specific actor_id, or all if not specified."""
|
|
if actor_id:
|
|
_prompt_cache.pop(actor_id, None)
|
|
else:
|
|
_prompt_cache.clear()
|