Spaces:
No application file
No application file
import json | |
import networkx as nx | |
from agents.agent import Agent | |
from locations.locations import Locations | |
from utils.text_generation import summarize_simulation | |
# Set default value for prompt_meta if not defined elsewhere | |
prompt_meta = '### Instruction:\n{}\n### Response:' | |
# Initialize global time and simulation variables | |
global_time = 0 | |
repeats = 5 | |
log_locations = False | |
log_actions = True | |
log_plans = False | |
log_ratings = False | |
log_memories = False | |
print_locations = True | |
print_actions = True | |
print_plans = True | |
print_ratings = True | |
print_memories = False | |
use_openai=True | |
# Start simulation loop | |
whole_simulation_output = "" | |
# Load town areas and people from JSON file | |
with open('simulation_config.json', 'r') as f: | |
town_data = json.load(f) | |
town_people = town_data['town_people'] | |
town_areas = town_data['town_areas'] | |
# Create world_graph | |
world_graph = nx.Graph() | |
last_town_area = None | |
for town_area in town_areas.keys(): | |
world_graph.add_node(town_area) | |
world_graph.add_edge(town_area, town_area) # Add an edge to itself | |
if last_town_area is not None: | |
world_graph.add_edge(town_area, last_town_area) | |
last_town_area = town_area | |
# Add the edge between the first and the last town areas to complete the cycle | |
world_graph.add_edge(list(town_areas.keys())[0], last_town_area) | |
# Initialize agents and locations | |
agents = [] | |
locations = Locations() | |
for name, description in town_people.items(): | |
starting_location = description['starting_location'] | |
agents.append(Agent(name, description['description'], starting_location, world_graph, use_openai)) | |
for name, description in town_areas.items(): | |
locations.add_location(name, description) | |
for repeat in range(repeats): | |
#log_output for one repeat | |
log_output = "" | |
print(f"====================== REPEAT {repeat} ======================\n") | |
log_output += f"====================== REPEAT {repeat} ======================\n" | |
if log_locations: | |
log_output += f"=== LOCATIONS AT START OF REPEAT {repeat} ===\n" | |
log_output += str(locations) + "\n" | |
if print_locations: | |
print(f"=== LOCATIONS AT START OF REPEAT {repeat} ===") | |
print(str(locations) + "\n") | |
# Plan actions for each agent | |
for agent in agents: | |
agent.plan(global_time, prompt_meta) | |
if log_plans: | |
log_output += f"{agent.name} plans: {agent.plans}\n" | |
if print_plans: | |
print(f"{agent.name} plans: {agent.plans}") | |
# Execute planned actions and update memories | |
for agent in agents: | |
# Execute action | |
action = agent.execute_action(agents, locations.get_location(agent.location), global_time, town_areas, prompt_meta) | |
if log_actions: | |
log_output += f"{agent.name} action: {action}\n" | |
if print_actions: | |
print(f"{agent.name} action: {action}") | |
# Update memories | |
for other_agent in agents: | |
if other_agent != agent: | |
memory = f'[Time: {global_time}. Person: {agent.name}. Memory: {action}]' | |
other_agent.memories.append(memory) | |
if log_memories: | |
log_output += f"{other_agent.name} remembers: {memory}\n" | |
if print_memories: | |
print(f"{other_agent.name} remembers: {memory}") | |
# Compress and rate memories for each agent | |
for agent in agents: | |
agent.compress_memories(global_time) | |
agent.rate_memories(locations, global_time, prompt_meta) | |
if log_ratings: | |
log_output += f"{agent.name} memory ratings: {agent.memory_ratings}\n" | |
if print_ratings: | |
print(f"{agent.name} memory ratings: {agent.memory_ratings}") | |
# Rate locations and determine where agents will go next | |
for agent in agents: | |
place_ratings = agent.rate_locations(locations, global_time, prompt_meta) | |
if log_ratings: | |
log_output += f"=== UPDATED LOCATION RATINGS {global_time} FOR {agent.name}===\n" | |
log_output += f"{agent.name} location ratings: {place_ratings}\n" | |
if print_ratings: | |
print(f"=== UPDATED LOCATION RATINGS {global_time} FOR {agent.name}===\n") | |
print(f"{agent.name} location ratings: {place_ratings}\n") | |
old_location = agent.location | |
new_location_name = place_ratings[0][0] | |
agent.move(new_location_name) | |
if print_locations: | |
log_output += f"=== UPDATED LOCATIONS AT TIME {global_time} FOR {agent.name}===\n" | |
log_output += f"{agent.name} moved from {old_location} to {new_location_name}\n" | |
if print_ratings: | |
print(f"=== UPDATED LOCATIONS AT TIME {global_time} FOR {agent.name}===\n") | |
print(f"{agent.name} moved from {old_location} to {new_location_name}\n") | |
print(f"----------------------- SUMMARY FOR REPEAT {repeat} -----------------------") | |
print(summarize_simulation(log_output=log_output)) | |
whole_simulation_output += log_output | |
# Increment time | |
global_time += 1 | |
# Write log output to file | |
with open('simulation_log.txt', 'w') as f: | |
f.write(whole_simulation_output) |