97 lines
4.3 KiB
Python
97 lines
4.3 KiB
Python
import os
|
|
import boto3
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
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)
|
|
|
|
# Dynamic block — injected fresh every call, never cached
|
|
chicago = ZoneInfo('America/Chicago')
|
|
now = datetime.now(chicago)
|
|
dt_str = now.strftime('%A %Y-%m-%d %H:%M:%S %Z')
|
|
time_block = (
|
|
f'## Current Time\n'
|
|
f'The current date and time is: {dt_str}\n\n'
|
|
f'When examining any other date or time value, calculate its distance from now '
|
|
f'(in seconds, minutes, hours, or days as appropriate) before drawing conclusions '
|
|
f'like "upcoming", "overdue", "recent", "just happened", or "a long time ago". '
|
|
f'Do this arithmetic explicitly — do not estimate or assume.'
|
|
)
|
|
|
|
parts = [base, time_block]
|
|
if user_context:
|
|
parts.append(f'## User\n{user_context}')
|
|
return '\n\n---\n\n'.join(parts)
|
|
|
|
|
|
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 = []
|
|
|
|
for fname in ['SOUL.md', 'STATUS.md']:
|
|
try:
|
|
obj = s3.get_object(Bucket=bucket, Key=fname)
|
|
content = obj['Body'].read().decode('utf-8')
|
|
if fname == 'STATUS.md':
|
|
parts.append(f'## Status — In Progress\n{content}')
|
|
else:
|
|
parts.append(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(
|
|
'## Memory\n'
|
|
'Your memory works through two layers:\n\n'
|
|
'**Conversation history (short-term):** AgentCore automatically loads your full '
|
|
'conversation history with this user at the start of each session. You have complete '
|
|
'context of everything discussed previously — no need to ask users to repeat themselves.\n\n'
|
|
'**Long-term facts (LTM):** Important facts extracted from past conversations are '
|
|
'retrieved and injected as context automatically. These are things like preferences, '
|
|
'setup details, names, and recurring topics the user has shared.\n\n'
|
|
'Guidelines:\n'
|
|
'- Never ask "what did we discuss last time?" — you already have the history.\n'
|
|
'- When a user shares something important (job interview, preference, key decision, '
|
|
'setup change), acknowledge it and trust it will be captured — do not ask if they '
|
|
'want you to remember it.\n'
|
|
'- If you notice a fact that seems important but may not be in LTM yet (e.g. a '
|
|
'deadline, a preference, a name), you may say "I\'ll keep that in mind" — but do '
|
|
'not ask permission or make a production of it.\n'
|
|
'- **In-progress tracking (STATUS.md):** When you start async work (CodeBuild job, '
|
|
'reminder, deployment, anything you need to check back on), update STATUS.md using '
|
|
"write_workspace_file('STATUS.md', content). Clear entries when complete. Check "
|
|
"STATUS.md at the start of sessions where Daniel asks 'what's happening' or 'any updates'."
|
|
)
|
|
parts.append('## Runtime\nRuntime: agent-claw | host=AgentCore | model=bedrock-claude-sonnet | channel=telegram | 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()
|