Persona Engagement Simulator
{isLoading && }
{!isLoading && results.length > 0 && (
)}
{!isLoading && results.length === 0 && (
Upload a CSV and run a simulation to see results.
)}
);
}
export default App;
```
### backend/agent.py
```python
# LETTA_API_KEY=sk-let-NGU5MzBjZWItODAzZC00MzE4LWIwODEtYTVmMTE1MzU5NTJlOjNkOGQzZGNmLWViNzMtNDVhNC1iNWM0LTI1ZDY3NzQ5OTIxNA==
```
### backend/debug_test.py
```python
import os
from letta_client import Letta, MessageCreate
from dotenv import load_dotenv
# Load environment variables
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '..', '.env'))
# Initialize client
api_key = os.getenv("LETTA_API_KEY")
client = Letta(token=api_key)
def test_simple_interaction():
"""Test a simple interaction with one agent to debug tool calling."""
# Get the first agent
agents = client.agents.list()
if not agents:
print("No agents found!")
return
agent = agents[0]
print(f"Testing with agent: {agent.name} ({agent.id})")
# Send a very simple message
simple_prompt = f"""
Please use the agent_like_ad tool with these parameters:
- agent_id: {agent.name}
- ad_id: test_ad
Just call the tool and tell me if it worked.
"""
print("\n=== Sending simple prompt ===")
print(simple_prompt)
try:
# Try streaming to see the full interaction
print("\n=== Streaming response ===")
stream = client.agents.messages.create_stream(
agent_id=agent.id,
messages=[MessageCreate(role="user", content=simple_prompt)]
)
print("Stream chunks:")
for i, chunk in enumerate(stream):
print(f" {i+1}. Type: {chunk.message_type}")
if hasattr(chunk, 'content') and chunk.content:
print(f" Content: {chunk.content}")
if hasattr(chunk, 'tool_call') and chunk.tool_call:
print(f" Tool: {chunk.tool_call.name}")
print(f" Args: {chunk.tool_call.arguments}")
if hasattr(chunk, 'tool_return') and chunk.tool_return:
print(f" Tool Return: {chunk.tool_return}")
except Exception as e:
print(f"Error in streaming: {e}")
if __name__ == "__main__":
test_simple_interaction()
```
### backend/tools_v2.py
```python
import os
import json
def agent_like_ad(agent_id: str, ad_id: str, **kwargs) -> str:
"""
Use this tool to express a 'like' for an advertisement.
Args:
agent_id (str): The ID of the agent performing the action.
ad_id (str): The unique identifier of the ad being liked.
Returns:
str: A JSON string confirming the action was successful.
"""
return json.dumps({"status": "success", "action": "like", "agent": agent_id, "ad": ad_id})
def agent_dislike_ad(agent_id: str, ad_id: str, **kwargs) -> str:
"""
Use this tool to express a 'dislike' for an advertisement.
Args:
agent_id (str): The ID of the agent performing the action.
ad_id (str): The unique identifier of the ad being disliked.
Returns:
str: A JSON string confirming the action was successful.
"""
return json.dumps({"status": "success", "action": "dislike", "agent": agent_id, "ad": ad_id})
def agent_comment_ad(agent_id: str, ad_id: str, comment_text: str, **kwargs) -> str:
"""
Use this tool to post a comment on an advertisement.
Args:
agent_id (str): The ID of the agent performing the action.
ad_id (str): The unique identifier of the ad for the comment.
comment_text (str): The content of the comment.
Returns:
str: A JSON string confirming the action was successful.
"""
return json.dumps({"status": "success", "action": "comment", "agent": agent_id, "ad": ad_id, "comment": comment_text})
def agent_repost_ad(agent_id: str, ad_id: str, repost_reason: str, **kwargs) -> str:
"""
Use this tool to repost an advertisement, similar to a retweet.
Args:
agent_id (str): The ID of the agent performing the action.
ad_id (str): The unique identifier of the ad being reposted.
repost_reason (str): The reason or commentary for the repost.
Returns:
str: A JSON string confirming the action was successful.
"""
return json.dumps({"status": "success", "action": "repost", "agent": agent_id, "ad": ad_id, "reason": repost_reason})
def agent_ignore_ad(agent_id: str, ad_id: str, **kwargs) -> str:
"""
Use this tool to ignore an advertisement without any engagement.
This represents scrolling past or dismissing an ad without interaction.
Args:
agent_id (str): The ID of the agent performing the action.
ad_id (str): The unique identifier of the ad being ignored.
Returns:
str: A JSON string confirming the action was successful.
"""
return json.dumps({"status": "success", "action": "ignore", "agent": agent_id, "ad": ad_id})
def read_shared_knowledge(**kwargs) -> str:
"""
Reads the shared knowledge base accessible to all agents.
Use this to understand collective trends or shared personality traits.
Returns:
str: The content of the shared knowledge file.
"""
SHARED_KNOWLEDGE_PATH = os.path.join(os.path.dirname(__file__), '..', 'data', 'shared_knowledge.txt')
try:
with open(SHARED_KNOWLEDGE_PATH, 'r') as f:
return f.read()
except FileNotFoundError:
return "No shared knowledge found."
def write_shared_knowledge(content: str, **kwargs) -> str:
"""
Writes or appends content to the shared knowledge base.
Use this to contribute to the collective memory or shared personality.
Args:
content (str): The information to add to the shared knowledge base.
Returns:
str: A confirmation message.
"""
SHARED_KNOWLEDGE_PATH = os.path.join(os.path.dirname(__file__), '..', 'data', 'shared_knowledge.txt')
with open(SHARED_KNOWLEDGE_PATH, 'a') as f:
f.write(f"\n{content}")
return "Shared knowledge updated successfully."
# A list of all tool functions for easy registration
ALL_TOOLS_V2 = [
agent_like_ad,
agent_dislike_ad,
agent_comment_ad,
agent_repost_ad,
agent_ignore_ad,
read_shared_knowledge,
write_shared_knowledge
]
```
### backend/simulation.py
```python
import os
import sys
import time
import asyncio
import json
import re
from letta_client import Letta, MessageCreate
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), '..', '.env'))
# --- Configuration ---
api_key = os.getenv("LETTA_API_KEY")
if not api_key:
raise ValueError("LETTA_API_KEY not found in .env file.")
CLIENT = Letta(token=api_key)
def extract_json_from_string(text: str) -> dict:
"""
Finds and parses the first valid JSON object within a string.
Handles cases where the JSON is embedded in other text.
"""
if not text or not text.strip():
print(f"Warning: Empty or whitespace-only response received")
return None
# Try to find JSON object enclosed in curly braces
# Look for the first { and the last } to handle nested objects
start_idx = text.find('{')
end_idx = text.rfind('}')
if start_idx == -1 or end_idx == -1 or start_idx >= end_idx:
print(f"Warning: No valid JSON structure found in text: '{text[:100]}...'")
return None
json_str = text[start_idx:end_idx + 1]
try:
return json.loads(json_str)
except json.JSONDecodeError as e:
print(f"Warning: Could not decode JSON from string: '{json_str[:100]}...', Error: {e}")
# Try to clean up common issues
# Remove any trailing commas before closing braces/brackets
cleaned = re.sub(r',(\s*[}\]])', r'\1', json_str)
try:
return json.loads(cleaned)
except json.JSONDecodeError:
print(f"Warning: Even cleaned JSON failed to parse: '{cleaned[:100]}...'")
return None
async def run_simulation_with_ad_copy(ad_copy: str):
"""Runs the simulation for all agents against a single ad, returning JSON results."""
ad_id = "user_provided_ad"
print(f"--- Starting Simulation for Ad: '{ad_id}' ---")
# 1. Get all available agents
try:
agents = CLIENT.agents.list()
if not agents:
print("No agents found. Please create them first.")
raise Exception("No agents available for simulation.")
except Exception as e:
print(f"Failed to connect to Letta server or fetch agent details. Error: {e}")
raise e
print(f"Found {len(agents)} agents. Presenting ad and collecting results...")
# 2. Present the ad to each agent concurrently
tasks = [run_agent_interaction(agent, ad_id, ad_copy) for agent in agents]
results = await asyncio.gather(*tasks)
# Filter out any None results from failed interactions
successful_results = [res for res in results if res]
print("\n--- Simulation Complete ---")
print(f"Successfully collected {len(successful_results)} results.")
return successful_results
async def run_agent_interaction(agent, ad_id: str, ad_content: str):
"""Presents an ad to a single agent and processes its response."""
print(f"\n-> Presenting ad to agent: {agent.name} ({agent.id})")
prompt = f"""
You are on a social media platform and you see the following ad.
Your name is {agent.name}. Your personality is stored in your 'persona' memory block.
Ad Content: "{ad_content}"
You must complete this task in TWO PHASES:
PHASE 1 - TAKE ACTIONS:
Based on your persona, use the provided tools to react to this ad. You can use one or more tools (e.g., like and comment).
PHASE 2 - PROVIDE ANALYSIS (MANDATORY):
After your tool calls, you MUST immediately provide a JSON analysis of your reaction.
Your JSON response must be a single line with no other text, starting with {{{{ and ending with }}}}.
**If you took multiple actions, for the "reaction" field in the JSON, choose the one that you feel is your PRIMARY reaction.** For example, if you liked and commented, and the comment is more significant, use "comment".
Required format:
{{"reaction": "primary_action", "confidence": 0-100, "reasoning": "why you reacted this way", "tags": ["keyword1", "keyword2"], "final_message": "your social media post"}}
The `reaction` value should be one of `like`, `dislike`, `comment`, `repost`, or `ignore`.
IMPORTANT: You MUST complete both phases. Do not stop after phase 1.
Example complete interaction:
1. [Agent uses tool: agent_like_ad]
2. [Agent uses tool: agent_comment_ad]
3. {{"reaction": "comment", "confidence": 90, "reasoning": "I liked it, but my main action is commenting to ask for more details.", "tags": ["eco", "fashion"], "final_message": "Love it! Can you provide more info on your ethical sourcing?"}}
"""
try:
# Send the prompt to the agent
print(f" - Sending prompt to {agent.name}...")
response = CLIENT.agents.messages.create_stream(
agent_id=agent.id,
messages=[MessageCreate(role="user", content=prompt)],
)
# Track tool calls and content
tool_calls = []
response_content = ""
for chunk in response:
if chunk.message_type == "assistant_message" and chunk.content:
response_content += chunk.content
elif chunk.message_type == "tool_call_message":
tool_name = chunk.tool_call.name
tool_calls.append(tool_name)
print(f" - Tool Call by {agent.name}: {tool_name}")
print(f" - Tool calls made: {tool_calls}")
print(f" - Raw response from {agent.name} (length: {len(response_content)}): '{response_content}'")
# If we got an empty response but tool calls were made, try to get a follow-up
if not response_content.strip() and tool_calls:
print(f" - Agent {agent.name} made tool calls but gave empty response. Requesting JSON...")
follow_up_stream = CLIENT.agents.messages.create_stream(
agent_id=agent.id,
messages=[MessageCreate(role="user", content="Please provide your JSON anal
[truncated — 2145 more characters]
```
[15 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]