remove-unused-params from agent.py

pull/1028/head
Kye Gomez 2 weeks ago
parent bb46bd9f94
commit 67b380dc8a

@ -30,6 +30,7 @@ try:
WikipediaPersonalityScraper,
MEPPersonalityProfile,
)
WIKIPEDIA_PERSONALITY_AVAILABLE = True
except ImportError:
WIKIPEDIA_PERSONALITY_AVAILABLE = False
@ -52,4 +53,4 @@ __all__ = [
"WikipediaPersonalityScraper",
"MEPPersonalityProfile",
"WIKIPEDIA_PERSONALITY_AVAILABLE",
]
]

@ -5,25 +5,21 @@ This script demonstrates the comprehensive democratic functionality of the EuroS
including bill introduction, committee work, parliamentary debates, and democratic voting.
"""
import json
import time
from datetime import datetime
# Import directly from the file
from euroswarm_parliament import (
EuroSwarmParliament,
VoteType,
ParliamentaryRole,
ParliamentaryMember
)
def demonstrate_parliament_initialization():
"""Demonstrate parliament initialization and basic functionality with cost optimization."""
print("\nEUROSWARM PARLIAMENT INITIALIZATION DEMONSTRATION (COST OPTIMIZED)")
print(
"\nEUROSWARM PARLIAMENT INITIALIZATION DEMONSTRATION (COST OPTIMIZED)"
)
print("=" * 60)
# Initialize the parliament with cost optimization
parliament = EuroSwarmParliament(
eu_data_file="EU.xml",
@ -35,487 +31,632 @@ def demonstrate_parliament_initialization():
enable_caching=True, # NEW: Enable response caching
batch_size=25, # NEW: Batch size for concurrent execution
budget_limit=100.0, # NEW: Budget limit in dollars
verbose=True
verbose=True,
)
print(f"Parliament initialized with {len(parliament.meps)} MEPs")
# Show parliament composition with cost stats
composition = parliament.get_parliament_composition()
print(f"\nPARLIAMENT COMPOSITION:")
print("\nPARLIAMENT COMPOSITION:")
print(f"Total MEPs: {composition['total_meps']}")
print(f"Loaded MEPs: {composition['loaded_meps']} (lazy loading active)")
print(f"\nCOST OPTIMIZATION:")
cost_stats = composition['cost_stats']
print(f"Budget Limit: ${cost_stats['budget_remaining'] + cost_stats['total_cost']:.2f}")
print(
f"Loaded MEPs: {composition['loaded_meps']} (lazy loading active)"
)
print("\nCOST OPTIMIZATION:")
cost_stats = composition["cost_stats"]
print(
f"Budget Limit: ${cost_stats['budget_remaining'] + cost_stats['total_cost']:.2f}"
)
print(f"Budget Used: ${cost_stats['total_cost']:.2f}")
print(f"Budget Remaining: ${cost_stats['budget_remaining']:.2f}")
print(f"Cache Hit Rate: {cost_stats['cache_hit_rate']:.1%}")
print(f"\nPOLITICAL GROUP DISTRIBUTION:")
for group, data in composition['political_groups'].items():
count = data['count']
percentage = data['percentage']
print("\nPOLITICAL GROUP DISTRIBUTION:")
for group, data in composition["political_groups"].items():
count = data["count"]
percentage = data["percentage"]
print(f" {group}: {count} MEPs ({percentage:.1f}%)")
print(f"\nCOMMITTEE LEADERSHIP:")
for committee_name, committee_data in composition['committees'].items():
chair = committee_data['chair']
print("\nCOMMITTEE LEADERSHIP:")
for committee_name, committee_data in composition[
"committees"
].items():
chair = committee_data["chair"]
if chair:
print(f" {committee_name}: {chair}")
return parliament
def demonstrate_individual_mep_interaction(parliament):
"""Demonstrate individual MEP interaction and personality."""
print("\nINDIVIDUAL MEP INTERACTION DEMONSTRATION")
print("=" * 60)
# Get a sample MEP
sample_mep_name = list(parliament.meps.keys())[0]
sample_mep = parliament.meps[sample_mep_name]
print(f"Sample MEP: {sample_mep.full_name}")
print(f"Country: {sample_mep.country}")
print(f"Political Group: {sample_mep.political_group}")
print(f"National Party: {sample_mep.national_party}")
print(f"Committees: {', '.join(sample_mep.committees)}")
print(f"Expertise Areas: {', '.join(sample_mep.expertise_areas)}")
# Test MEP agent interaction
if sample_mep.agent:
test_prompt = "What are your views on European integration and how do you approach cross-border cooperation?"
print(f"\nMEP Response to: '{test_prompt}'")
print("-" * 50)
try:
response = sample_mep.agent.run(test_prompt)
print(response[:500] + "..." if len(response) > 500 else response)
print(
response[:500] + "..."
if len(response) > 500
else response
)
except Exception as e:
print(f"Error getting MEP response: {e}")
def demonstrate_committee_work(parliament):
"""Demonstrate committee work and hearings."""
print("\nCOMMITTEE WORK DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[0]
# Create a test bill
bill = parliament.introduce_bill(
title="European Digital Rights and Privacy Protection Act",
description="Comprehensive legislation to strengthen digital rights, enhance privacy protection, and establish clear guidelines for data handling across the European Union.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Legal Affairs",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Bill: {bill.title}")
print(f"Committee: {bill.committee}")
print(f"Sponsor: {bill.sponsor}")
# Conduct committee hearing
print(f"\nCONDUCTING COMMITTEE HEARING...")
hearing_result = parliament.conduct_committee_hearing(bill.committee, bill)
print("\nCONDUCTING COMMITTEE HEARING...")
hearing_result = parliament.conduct_committee_hearing(
bill.committee, bill
)
print(f"Committee: {hearing_result['committee']}")
print(f"Participants: {len(hearing_result['participants'])} MEPs")
print(f"Recommendation: {hearing_result['recommendations']['recommendation']}")
print(f"Support: {hearing_result['recommendations']['support_percentage']:.1f}%")
print(f"Oppose: {hearing_result['recommendations']['oppose_percentage']:.1f}%")
print(f"Amend: {hearing_result['recommendations']['amend_percentage']:.1f}%")
print(
f"Recommendation: {hearing_result['recommendations']['recommendation']}"
)
print(
f"Support: {hearing_result['recommendations']['support_percentage']:.1f}%"
)
print(
f"Oppose: {hearing_result['recommendations']['oppose_percentage']:.1f}%"
)
print(
f"Amend: {hearing_result['recommendations']['amend_percentage']:.1f}%"
)
def demonstrate_parliamentary_debate(parliament):
"""Demonstrate parliamentary debate functionality."""
print("\nPARLIAMENTARY DEBATE DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[1]
# Create a test bill
bill = parliament.introduce_bill(
title="European Green Deal Implementation Act",
description="Legislation to implement the European Green Deal, including carbon neutrality targets, renewable energy investments, and sustainable development measures.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Environment, Public Health and Food Safety",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Bill: {bill.title}")
print(f"Description: {bill.description}")
# Conduct parliamentary debate
print(f"\nCONDUCTING PARLIAMENTARY DEBATE...")
debate_result = parliament.conduct_parliamentary_debate(bill, max_speakers=10)
print(f"Debate Participants: {len(debate_result['participants'])} MEPs")
print(f"Debate Analysis:")
print(f" Support: {debate_result['analysis']['support_count']} speakers ({debate_result['analysis']['support_percentage']:.1f}%)")
print(f" Oppose: {debate_result['analysis']['oppose_count']} speakers ({debate_result['analysis']['oppose_percentage']:.1f}%)")
print(f" Neutral: {debate_result['analysis']['neutral_count']} speakers ({debate_result['analysis']['neutral_percentage']:.1f}%)")
print("\nCONDUCTING PARLIAMENTARY DEBATE...")
debate_result = parliament.conduct_parliamentary_debate(
bill, max_speakers=10
)
print(
f"Debate Participants: {len(debate_result['participants'])} MEPs"
)
print("Debate Analysis:")
print(
f" Support: {debate_result['analysis']['support_count']} speakers ({debate_result['analysis']['support_percentage']:.1f}%)"
)
print(
f" Oppose: {debate_result['analysis']['oppose_count']} speakers ({debate_result['analysis']['oppose_percentage']:.1f}%)"
)
print(
f" Neutral: {debate_result['analysis']['neutral_count']} speakers ({debate_result['analysis']['neutral_percentage']:.1f}%)"
)
def demonstrate_democratic_voting(parliament):
"""Demonstrate democratic voting functionality."""
print("\nDEMOCRATIC VOTING DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[2]
# Create a test bill
bill = parliament.introduce_bill(
title="European Social Rights and Labor Protection Act",
description="Legislation to strengthen social rights, improve labor conditions, and ensure fair treatment of workers across the European Union.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Employment and Social Affairs",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Bill: {bill.title}")
print(f"Sponsor: {bill.sponsor}")
# Conduct democratic vote
print(f"\nCONDUCTING DEMOCRATIC VOTE...")
print("\nCONDUCTING DEMOCRATIC VOTE...")
vote_result = parliament.conduct_democratic_vote(bill)
# Calculate percentages
total_votes = vote_result.votes_for + vote_result.votes_against + vote_result.abstentions
in_favor_percentage = (vote_result.votes_for / total_votes * 100) if total_votes > 0 else 0
against_percentage = (vote_result.votes_against / total_votes * 100) if total_votes > 0 else 0
abstentions_percentage = (vote_result.abstentions / total_votes * 100) if total_votes > 0 else 0
print(f"Vote Results:")
total_votes = (
vote_result.votes_for
+ vote_result.votes_against
+ vote_result.abstentions
)
in_favor_percentage = (
(vote_result.votes_for / total_votes * 100)
if total_votes > 0
else 0
)
against_percentage = (
(vote_result.votes_against / total_votes * 100)
if total_votes > 0
else 0
)
abstentions_percentage = (
(vote_result.abstentions / total_votes * 100)
if total_votes > 0
else 0
)
print("Vote Results:")
print(f" Total Votes: {total_votes}")
print(f" In Favor: {vote_result.votes_for} ({in_favor_percentage:.1f}%)")
print(f" Against: {vote_result.votes_against} ({against_percentage:.1f}%)")
print(f" Abstentions: {vote_result.abstentions} ({abstentions_percentage:.1f}%)")
print(
f" In Favor: {vote_result.votes_for} ({in_favor_percentage:.1f}%)"
)
print(
f" Against: {vote_result.votes_against} ({against_percentage:.1f}%)"
)
print(
f" Abstentions: {vote_result.abstentions} ({abstentions_percentage:.1f}%)"
)
print(f" Result: {vote_result.result.value}")
# Show political group breakdown if available
if hasattr(vote_result, 'group_votes') and vote_result.group_votes:
print(f"\nPOLITICAL GROUP BREAKDOWN:")
if (
hasattr(vote_result, "group_votes")
and vote_result.group_votes
):
print("\nPOLITICAL GROUP BREAKDOWN:")
for group, votes in vote_result.group_votes.items():
print(f" {group}: {votes['in_favor']}/{votes['total']} in favor ({votes['percentage']:.1f}%)")
print(
f" {group}: {votes['in_favor']}/{votes['total']} in favor ({votes['percentage']:.1f}%)"
)
else:
print(f"\nIndividual votes recorded: {len(vote_result.individual_votes)} MEPs")
print(
f"\nIndividual votes recorded: {len(vote_result.individual_votes)} MEPs"
)
def demonstrate_complete_democratic_session(parliament):
"""Demonstrate a complete democratic parliamentary session."""
print("\nCOMPLETE DEMOCRATIC SESSION DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[3]
# Run complete session
session_result = parliament.run_democratic_session(
bill_title="European Innovation and Technology Advancement Act",
bill_description="Comprehensive legislation to promote innovation, support technology startups, and establish Europe as a global leader in digital transformation and technological advancement.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Industry, Research and Energy",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Session Results:")
print("Session Results:")
print(f" Bill: {session_result['bill'].title}")
print(f" Committee Hearing: {session_result['hearing']['recommendations']['recommendation']}")
print(f" Debate Participants: {len(session_result['debate']['participants'])} MEPs")
print(
f" Committee Hearing: {session_result['hearing']['recommendations']['recommendation']}"
)
print(
f" Debate Participants: {len(session_result['debate']['participants'])} MEPs"
)
print(f" Final Vote: {session_result['vote']['result']}")
print(f" Vote Margin: {session_result['vote']['in_favor_percentage']:.1f}% in favor")
print(
f" Vote Margin: {session_result['vote']['in_favor_percentage']:.1f}% in favor"
)
def demonstrate_political_analysis(parliament):
"""Demonstrate political analysis and voting prediction."""
print("\nPOLITICAL ANALYSIS DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[4]
# Create a test bill
bill = parliament.introduce_bill(
title="European Climate Action and Sustainability Act",
description="Comprehensive climate action legislation including carbon pricing, renewable energy targets, and sustainable development measures.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Environment, Public Health and Food Safety",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Bill: {bill.title}")
print(f"Sponsor: {bill.sponsor}")
# Analyze political landscape
analysis = parliament.analyze_political_landscape(bill)
print(f"\nPOLITICAL LANDSCAPE ANALYSIS:")
print("\nPOLITICAL LANDSCAPE ANALYSIS:")
print(f" Overall Support: {analysis['overall_support']:.1f}%")
print(f" Opposition: {analysis['opposition']:.1f}%")
print(f" Uncertainty: {analysis['uncertainty']:.1f}%")
print(f"\nPOLITICAL GROUP ANALYSIS:")
for group, data in analysis['group_analysis'].items():
print(f" {group}: {data['support']:.1f}% support, {data['opposition']:.1f}% opposition")
print("\nPOLITICAL GROUP ANALYSIS:")
for group, data in analysis["group_analysis"].items():
print(
f" {group}: {data['support']:.1f}% support, {data['opposition']:.1f}% opposition"
)
def demonstrate_hierarchical_democratic_voting(parliament):
"""Demonstrate hierarchical democratic voting with political group boards."""
print("\nHIERARCHICAL DEMOCRATIC VOTING DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[5]
# Create a test bill
bill = parliament.introduce_bill(
title="European Climate Action and Sustainability Act",
description="Comprehensive climate action legislation including carbon pricing, renewable energy targets, and sustainable development measures.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Environment, Public Health and Food Safety",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Bill: {bill.title}")
print(f"Sponsor: {bill.sponsor}")
# Conduct hierarchical vote
print(f"\nCONDUCTING HIERARCHICAL DEMOCRATIC VOTE...")
hierarchical_result = parliament.conduct_hierarchical_democratic_vote(bill)
print(f"Hierarchical Vote Results:")
print("\nCONDUCTING HIERARCHICAL DEMOCRATIC VOTE...")
hierarchical_result = (
parliament.conduct_hierarchical_democratic_vote(bill)
)
print("Hierarchical Vote Results:")
print(f" Total Votes: {hierarchical_result['total_votes']}")
print(f" In Favor: {hierarchical_result['in_favor']} ({hierarchical_result['in_favor_percentage']:.1f}%)")
print(f" Against: {hierarchical_result['against']} ({hierarchical_result['against_percentage']:.1f}%)")
print(
f" In Favor: {hierarchical_result['in_favor']} ({hierarchical_result['in_favor_percentage']:.1f}%)"
)
print(
f" Against: {hierarchical_result['against']} ({hierarchical_result['against_percentage']:.1f}%)"
)
print(f" Result: {hierarchical_result['result']}")
print(f"\nPOLITICAL GROUP BOARD DECISIONS:")
for group, decision in hierarchical_result['group_decisions'].items():
print(f" {group}: {decision['decision']} ({decision['confidence']:.1f}% confidence)")
print("\nPOLITICAL GROUP BOARD DECISIONS:")
for group, decision in hierarchical_result[
"group_decisions"
].items():
print(
f" {group}: {decision['decision']} ({decision['confidence']:.1f}% confidence)"
)
def demonstrate_complete_hierarchical_session(parliament):
"""Demonstrate a complete hierarchical democratic session."""
print("\nCOMPLETE HIERARCHICAL DEMOCRATIC SESSION DEMONSTRATION")
print("=" * 60)
# Get a real MEP as sponsor
sponsor = list(parliament.meps.keys())[6]
# Run complete hierarchical session
session_result = parliament.run_hierarchical_democratic_session(
bill_title="European Climate Action and Sustainability Act",
bill_description="Comprehensive climate action legislation including carbon pricing, renewable energy targets, and sustainable development measures.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Environment, Public Health and Food Safety",
sponsor=sponsor
sponsor=sponsor,
)
print(f"Hierarchical Session Results:")
print("Hierarchical Session Results:")
print(f" Bill: {session_result['bill'].title}")
print(f" Committee Hearing: {session_result['hearing']['recommendations']['recommendation']}")
print(f" Debate Participants: {len(session_result['debate']['participants'])} MEPs")
print(
f" Committee Hearing: {session_result['hearing']['recommendations']['recommendation']}"
)
print(
f" Debate Participants: {len(session_result['debate']['participants'])} MEPs"
)
print(f" Final Vote: {session_result['vote']['result']}")
print(f" Vote Margin: {session_result['vote']['in_favor_percentage']:.1f}% in favor")
print(
f" Vote Margin: {session_result['vote']['in_favor_percentage']:.1f}% in favor"
)
def demonstrate_wikipedia_personalities(parliament):
"""Demonstrate the Wikipedia personality system for realistic MEP behavior."""
print("\nWIKIPEDIA PERSONALITY SYSTEM DEMONSTRATION")
print("=" * 60)
# Check if Wikipedia personalities are available
if not parliament.enable_wikipedia_personalities:
print("Wikipedia personality system not available")
print("To enable: Install required dependencies and run Wikipedia scraper")
print(
"To enable: Install required dependencies and run Wikipedia scraper"
)
return
print(f"Wikipedia personality system enabled")
print(f"Loaded {len(parliament.personality_profiles)} personality profiles")
print("Wikipedia personality system enabled")
print(
f"Loaded {len(parliament.personality_profiles)} personality profiles"
)
# Show sample personality profiles
print(f"\nSAMPLE PERSONALITY PROFILES:")
print("\nSAMPLE PERSONALITY PROFILES:")
print("-" * 40)
sample_count = 0
for mep_name, profile in parliament.personality_profiles.items():
if sample_count >= 3: # Show only 3 samples
break
print(f"\n{mep_name}")
print(f" Wikipedia URL: {profile.wikipedia_url if profile.wikipedia_url else 'Not available'}")
print(f" Summary: {profile.summary[:200]}..." if profile.summary else "No summary available")
print(f" Political Views: {profile.political_views[:150]}..." if profile.political_views else "Based on party alignment")
print(f" Policy Focus: {profile.policy_focus[:150]}..." if profile.policy_focus else "General parliamentary work")
print(f" Achievements: {profile.achievements[:150]}..." if profile.achievements else "Parliamentary service")
print(
f" Wikipedia URL: {profile.wikipedia_url if profile.wikipedia_url else 'Not available'}"
)
print(
f" Summary: {profile.summary[:200]}..."
if profile.summary
else "No summary available"
)
print(
f" Political Views: {profile.political_views[:150]}..."
if profile.political_views
else "Based on party alignment"
)
print(
f" Policy Focus: {profile.policy_focus[:150]}..."
if profile.policy_focus
else "General parliamentary work"
)
print(
f" Achievements: {profile.achievements[:150]}..."
if profile.achievements
else "Parliamentary service"
)
print(f" Last Updated: {profile.last_updated}")
sample_count += 1
# Demonstrate personality-driven voting
print(f"\nPERSONALITY-DRIVEN VOTING DEMONSTRATION:")
print("\nPERSONALITY-DRIVEN VOTING DEMONSTRATION:")
print("-" * 50)
# Create a test bill that would trigger different personality responses
bill = parliament.introduce_bill(
title="European Climate Action and Green Technology Investment Act",
description="Comprehensive legislation to accelerate Europe's transition to renewable energy, including massive investments in green technology, carbon pricing mechanisms, and support for affected industries and workers.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Environment",
sponsor="Climate Action Leader"
sponsor="Climate Action Leader",
)
print(f"Bill: {bill.title}")
print(f"Description: {bill.description}")
# Show how different MEPs with Wikipedia personalities would respond
print(f"\nPERSONALITY-BASED RESPONSES:")
print("\nPERSONALITY-BASED RESPONSES:")
print("-" * 40)
sample_meps = list(parliament.personality_profiles.keys())[:3]
for mep_name in sample_meps:
mep = parliament.meps.get(mep_name)
profile = parliament.personality_profiles.get(mep_name)
if mep and profile:
print(f"\n{mep_name} ({mep.political_group})")
# Show personality influence
if profile.political_views:
print(f" Political Views: {profile.political_views[:100]}...")
print(
f" Political Views: {profile.political_views[:100]}..."
)
if profile.policy_focus:
print(f" Policy Focus: {profile.policy_focus[:100]}...")
print(
f" Policy Focus: {profile.policy_focus[:100]}..."
)
# Predict voting behavior based on personality
if "environment" in profile.policy_focus.lower() or "climate" in profile.political_views.lower():
if (
"environment" in profile.policy_focus.lower()
or "climate" in profile.political_views.lower()
):
predicted_vote = "LIKELY SUPPORT"
reasoning = "Environmental policy focus and climate advocacy"
elif "economic" in profile.policy_focus.lower() or "business" in profile.political_views.lower():
reasoning = (
"Environmental policy focus and climate advocacy"
)
elif (
"economic" in profile.policy_focus.lower()
or "business" in profile.political_views.lower()
):
predicted_vote = "LIKELY OPPOSE"
reasoning = "Economic concerns about investment costs"
else:
predicted_vote = "UNCERTAIN"
reasoning = "Mixed considerations based on party alignment"
reasoning = (
"Mixed considerations based on party alignment"
)
print(f" Predicted Vote: {predicted_vote}")
print(f" Reasoning: {reasoning}")
# Demonstrate scraping functionality
print(f"\nWIKIPEDIA SCRAPING CAPABILITIES:")
print("\nWIKIPEDIA SCRAPING CAPABILITIES:")
print("-" * 50)
print("Can scrape Wikipedia data for all 717 MEPs")
print("Extracts political views, career history, and achievements")
print(
"Extracts political views, career history, and achievements"
)
print("Creates detailed personality profiles in JSON format")
print("Integrates real personality data into AI agent system prompts")
print(
"Integrates real personality data into AI agent system prompts"
)
print("Enables realistic, personality-driven voting behavior")
print("Respectful API usage with configurable delays")
print(f"\nTo scrape all MEP personalities:")
print("\nTo scrape all MEP personalities:")
print(" parliament.scrape_wikipedia_personalities(delay=1.0)")
print(" # This will create personality profiles for all 717 MEPs")
print(
" # This will create personality profiles for all 717 MEPs"
)
print(" # Profiles are saved in 'mep_personalities/' directory")
def demonstrate_optimized_parliamentary_session(parliament):
"""Demonstrate cost-optimized parliamentary session."""
print("\nCOST-OPTIMIZED PARLIAMENTARY SESSION DEMONSTRATION")
print("=" * 60)
# Run optimized session with cost limit
session_result = parliament.run_optimized_parliamentary_session(
bill_title="European Digital Rights and Privacy Protection Act",
bill_description="Comprehensive legislation to strengthen digital rights, enhance privacy protection, and establish clear guidelines for data handling across the European Union.",
bill_type=VoteType.ORDINARY_LEGISLATIVE_PROCEDURE,
committee="Legal Affairs",
max_cost=25.0 # Max $25 for this session
)
print(f"Session Results:")
print(f" Bill: {session_result['session_summary']['bill_title']}")
print(f" Final Outcome: {session_result['session_summary']['final_outcome']}")
print(f" Total Cost: ${session_result['session_summary']['total_cost']:.2f}")
print(f" Budget Remaining: ${session_result['cost_stats']['budget_remaining']:.2f}")
max_cost=25.0, # Max $25 for this session
)
print("Session Results:")
print(
f" Bill: {session_result['session_summary']['bill_title']}"
)
print(
f" Final Outcome: {session_result['session_summary']['final_outcome']}"
)
print(
f" Total Cost: ${session_result['session_summary']['total_cost']:.2f}"
)
print(
f" Budget Remaining: ${session_result['cost_stats']['budget_remaining']:.2f}"
)
# Show detailed cost statistics
cost_stats = parliament.get_cost_statistics()
print(f"\nDETAILED COST STATISTICS:")
print("\nDETAILED COST STATISTICS:")
print(f" Total Tokens Used: {cost_stats['total_tokens']:,}")
print(f" Requests Made: {cost_stats['requests_made']}")
print(f" Cache Hits: {cost_stats['cache_hits']}")
print(f" Cache Hit Rate: {cost_stats['cache_hit_rate']:.1%}")
print(f" Loading Efficiency: {cost_stats['loading_efficiency']:.1%}")
print(
f" Loading Efficiency: {cost_stats['loading_efficiency']:.1%}"
)
print(f" Cache Size: {cost_stats['cache_size']} entries")
return session_result
def main():
"""Main demonstration function."""
print("EUROSWARM PARLIAMENT - COST OPTIMIZED DEMONSTRATION")
print("=" * 60)
print("This demonstration shows the EuroSwarm Parliament with cost optimization features:")
print(
"This demonstration shows the EuroSwarm Parliament with cost optimization features:"
)
print("• Lazy loading of MEP agents (only create when needed)")
print("• Response caching (avoid repeated API calls)")
print("• Batch processing (control memory and cost)")
print("• Budget controls (hard limits on spending)")
print("• Cost tracking (real-time monitoring)")
# Initialize parliament with cost optimization
parliament = demonstrate_parliament_initialization()
# Demonstrate individual MEP interaction (will trigger lazy loading)
demonstrate_individual_mep_interaction(parliament)
# Demonstrate committee work with cost optimization
demonstrate_committee_work(parliament)
# Demonstrate parliamentary debate with cost optimization
demonstrate_parliamentary_debate(parliament)
# Demonstrate democratic voting with cost optimization
demonstrate_democratic_voting(parliament)
# Demonstrate political analysis with cost optimization
demonstrate_political_analysis(parliament)
# Demonstrate optimized parliamentary session
demonstrate_optimized_parliamentary_session(parliament)
# Show final cost statistics
final_stats = parliament.get_cost_statistics()
print(f"\nFINAL COST STATISTICS:")
print("\nFINAL COST STATISTICS:")
print(f"Total Cost: ${final_stats['total_cost']:.2f}")
print(f"Budget Remaining: ${final_stats['budget_remaining']:.2f}")
print(f"Cache Hit Rate: {final_stats['cache_hit_rate']:.1%}")
print(f"Loading Efficiency: {final_stats['loading_efficiency']:.1%}")
print(f"\n✅ COST OPTIMIZATION DEMONSTRATION COMPLETED!")
print(f"✅ EuroSwarm Parliament now supports cost-effective large-scale simulations")
print(f"✅ Lazy loading: {final_stats['loaded_meps']}/{final_stats['total_meps']} MEPs loaded")
print(
f"Loading Efficiency: {final_stats['loading_efficiency']:.1%}"
)
print("\n✅ COST OPTIMIZATION DEMONSTRATION COMPLETED!")
print(
"✅ EuroSwarm Parliament now supports cost-effective large-scale simulations"
)
print(
f"✅ Lazy loading: {final_stats['loaded_meps']}/{final_stats['total_meps']} MEPs loaded"
)
print(f"✅ Caching: {final_stats['cache_hit_rate']:.1%} hit rate")
print(f"✅ Budget control: ${final_stats['total_cost']:.2f} spent of ${final_stats['budget_remaining'] + final_stats['total_cost']:.2f} budget")
print(
f"✅ Budget control: ${final_stats['total_cost']:.2f} spent of ${final_stats['budget_remaining'] + final_stats['total_cost']:.2f} budget"
)
if __name__ == "__main__":
main()
main()

@ -5,63 +5,76 @@ Test script to verify mass agent template can process more than 500 agents.
from mass_agent_template import MassAgentTemplate
def test_mass_agents():
print("Testing Mass Agent Template - Processing More Than 50 Agents")
print(
"Testing Mass Agent Template - Processing More Than 50 Agents"
)
print("=" * 60)
# Initialize template with 200 agents
template = MassAgentTemplate(
agent_count=200,
budget_limit=50.0,
batch_size=25,
verbose=True
verbose=True,
)
print(f"Initialized with {len(template.agents)} agents")
print(f"Budget limit: ${template.cost_tracker.budget_limit}")
# Test processing 100 agents
print(f"\nTesting with 100 agents...")
print("\nTesting with 100 agents...")
result = template.run_mass_task(
"What is the most important skill for your role?",
agent_count=100
agent_count=100,
)
print(f"Results:")
print("Results:")
print(f" Agents processed: {len(result['agents_used'])}")
print(f" Cost: ${result['cost_stats']['total_cost']:.4f}")
print(f" Budget remaining: ${result['cost_stats']['budget_remaining']:.2f}")
print(
f" Budget remaining: ${result['cost_stats']['budget_remaining']:.2f}"
)
print(f" Cached: {result.get('cached', False)}")
# Test processing 150 agents
print(f"\nTesting with 150 agents...")
print("\nTesting with 150 agents...")
result2 = template.run_mass_task(
"Describe your approach to problem-solving",
agent_count=150
"Describe your approach to problem-solving", agent_count=150
)
print(f"Results:")
print("Results:")
print(f" Agents processed: {len(result2['agents_used'])}")
print(f" Cost: ${result2['cost_stats']['total_cost']:.4f}")
print(f" Budget remaining: ${result2['cost_stats']['budget_remaining']:.2f}")
print(
f" Budget remaining: ${result2['cost_stats']['budget_remaining']:.2f}"
)
print(f" Cached: {result2.get('cached', False)}")
# Show final stats
final_stats = template.get_system_stats()
print(f"\nFinal Statistics:")
print("\nFinal Statistics:")
print(f" Total agents: {final_stats['total_agents']}")
print(f" Loaded agents: {final_stats['loaded_agents']}")
print(f" Total cost: ${final_stats['cost_stats']['total_cost']:.4f}")
print(f" Budget remaining: ${final_stats['cost_stats']['budget_remaining']:.2f}")
print(
f" Total cost: ${final_stats['cost_stats']['total_cost']:.4f}"
)
print(
f" Budget remaining: ${final_stats['cost_stats']['budget_remaining']:.2f}"
)
# Success criteria
total_processed = len(result['agents_used']) + len(result2['agents_used'])
total_processed = len(result["agents_used"]) + len(
result2["agents_used"]
)
print(f"\nTotal agents processed: {total_processed}")
if total_processed > 50:
print("✅ SUCCESS: Template processed more than 50 agents!")
else:
print("❌ FAILURE: Template still limited to 50 agents")
if __name__ == "__main__":
test_mass_agents()
test_mass_agents()

@ -14,14 +14,13 @@ from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
import requests
from loguru import logger
import xml.etree.ElementTree as ET
@dataclass
class MEPPersonalityProfile:
"""
Comprehensive personality profile for an MEP based on Wikipedia data.
Attributes:
full_name: Full name of the MEP
mep_id: Unique MEP identifier
@ -47,7 +46,7 @@ class MEPPersonalityProfile:
social_media: Social media presence
last_updated: When the profile was last updated
"""
full_name: str
mep_id: str
wikipedia_url: Optional[str] = None
@ -77,11 +76,15 @@ class WikipediaPersonalityScraper:
"""
Scraper for gathering Wikipedia personality data for MEPs.
"""
def __init__(self, output_dir: str = "mep_personalities", verbose: bool = True):
def __init__(
self,
output_dir: str = "mep_personalities",
verbose: bool = True,
):
"""
Initialize the Wikipedia personality scraper.
Args:
output_dir: Directory to store personality profiles
verbose: Enable verbose logging
@ -89,61 +92,81 @@ class WikipediaPersonalityScraper:
self.output_dir = output_dir
self.verbose = verbose
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'EuroSwarm Parliament Personality Scraper/1.0 (https://github.com/swarms-democracy)'
})
self.session.headers.update(
{
"User-Agent": "EuroSwarm Parliament Personality Scraper/1.0 (https://github.com/swarms-democracy)"
}
)
# Create output directory
os.makedirs(output_dir, exist_ok=True)
if verbose:
logger.info(f"Wikipedia Personality Scraper initialized. Output directory: {output_dir}")
logger.info(
f"Wikipedia Personality Scraper initialized. Output directory: {output_dir}"
)
def extract_mep_data_from_xml(self, xml_file: str = "EU.xml") -> List[Dict[str, str]]:
def extract_mep_data_from_xml(
self, xml_file: str = "EU.xml"
) -> List[Dict[str, str]]:
"""
Extract MEP data from EU.xml file.
Args:
xml_file: Path to EU.xml file
Returns:
List of MEP data dictionaries
"""
meps = []
try:
with open(xml_file, 'r', encoding='utf-8') as f:
with open(xml_file, "r", encoding="utf-8") as f:
content = f.read()
# Use regex to extract MEP data
mep_pattern = r'<mep>\s*<fullName>(.*?)</fullName>\s*<country>(.*?)</country>\s*<politicalGroup>(.*?)</politicalGroup>\s*<id>(.*?)</id>\s*<nationalPoliticalGroup>(.*?)</nationalPoliticalGroup>\s*</mep>'
mep_pattern = r"<mep>\s*<fullName>(.*?)</fullName>\s*<country>(.*?)</country>\s*<politicalGroup>(.*?)</politicalGroup>\s*<id>(.*?)</id>\s*<nationalPoliticalGroup>(.*?)</nationalPoliticalGroup>\s*</mep>"
mep_matches = re.findall(mep_pattern, content, re.DOTALL)
for full_name, country, political_group, mep_id, national_party in mep_matches:
meps.append({
'full_name': full_name.strip(),
'country': country.strip(),
'political_group': political_group.strip(),
'mep_id': mep_id.strip(),
'national_party': national_party.strip()
})
for (
full_name,
country,
political_group,
mep_id,
national_party,
) in mep_matches:
meps.append(
{
"full_name": full_name.strip(),
"country": country.strip(),
"political_group": political_group.strip(),
"mep_id": mep_id.strip(),
"national_party": national_party.strip(),
}
)
if self.verbose:
logger.info(f"Extracted {len(meps)} MEPs from {xml_file}")
logger.info(
f"Extracted {len(meps)} MEPs from {xml_file}"
)
except Exception as e:
logger.error(f"Error extracting MEP data from {xml_file}: {e}")
logger.error(
f"Error extracting MEP data from {xml_file}: {e}"
)
return meps
def search_wikipedia_page(self, mep_name: str, country: str) -> Optional[str]:
def search_wikipedia_page(
self, mep_name: str, country: str
) -> Optional[str]:
"""
Search for a Wikipedia page for an MEP.
Args:
mep_name: Full name of the MEP
country: Country of the MEP
Returns:
Wikipedia page title if found, None otherwise
"""
@ -151,48 +174,56 @@ class WikipediaPersonalityScraper:
# Search for the MEP on Wikipedia
search_url = "https://en.wikipedia.org/w/api.php"
search_params = {
'action': 'query',
'format': 'json',
'list': 'search',
'srsearch': f'"{mep_name}" {country}',
'srlimit': 5,
'srnamespace': 0
"action": "query",
"format": "json",
"list": "search",
"srsearch": f'"{mep_name}" {country}',
"srlimit": 5,
"srnamespace": 0,
}
response = self.session.get(search_url, params=search_params)
response = self.session.get(
search_url, params=search_params
)
response.raise_for_status()
data = response.json()
search_results = data.get('query', {}).get('search', [])
search_results = data.get("query", {}).get("search", [])
if search_results:
# Return the first result
return search_results[0]['title']
return search_results[0]["title"]
# Try alternative search without quotes
search_params['srsearch'] = f'{mep_name} {country}'
response = self.session.get(search_url, params=search_params)
search_params["srsearch"] = f"{mep_name} {country}"
response = self.session.get(
search_url, params=search_params
)
response.raise_for_status()
data = response.json()
search_results = data.get('query', {}).get('search', [])
search_results = data.get("query", {}).get("search", [])
if search_results:
return search_results[0]['title']
return search_results[0]["title"]
except Exception as e:
if self.verbose:
logger.warning(f"Error searching Wikipedia for {mep_name}: {e}")
logger.warning(
f"Error searching Wikipedia for {mep_name}: {e}"
)
return None
def get_wikipedia_content(self, page_title: str) -> Optional[Dict[str, Any]]:
def get_wikipedia_content(
self, page_title: str
) -> Optional[Dict[str, Any]]:
"""
Get Wikipedia content for a specific page.
Args:
page_title: Wikipedia page title
Returns:
Dictionary containing page content and metadata
"""
@ -200,376 +231,451 @@ class WikipediaPersonalityScraper:
# Get page content
content_url = "https://en.wikipedia.org/w/api.php"
content_params = {
'action': 'query',
'format': 'json',
'titles': page_title,
'prop': 'extracts|info|categories',
'exintro': True,
'explaintext': True,
'inprop': 'url',
'cllimit': 50
"action": "query",
"format": "json",
"titles": page_title,
"prop": "extracts|info|categories",
"exintro": True,
"explaintext": True,
"inprop": "url",
"cllimit": 50,
}
response = self.session.get(content_url, params=content_params)
response = self.session.get(
content_url, params=content_params
)
response.raise_for_status()
data = response.json()
pages = data.get('query', {}).get('pages', {})
pages = data.get("query", {}).get("pages", {})
if pages:
page_id = list(pages.keys())[0]
page_data = pages[page_id]
return {
'title': page_data.get('title', ''),
'extract': page_data.get('extract', ''),
'url': page_data.get('fullurl', ''),
'categories': [cat['title'] for cat in page_data.get('categories', [])],
'pageid': page_data.get('pageid', ''),
'length': page_data.get('length', 0)
"title": page_data.get("title", ""),
"extract": page_data.get("extract", ""),
"url": page_data.get("fullurl", ""),
"categories": [
cat["title"]
for cat in page_data.get("categories", [])
],
"pageid": page_data.get("pageid", ""),
"length": page_data.get("length", 0),
}
except Exception as e:
if self.verbose:
logger.warning(f"Error getting Wikipedia content for {page_title}: {e}")
logger.warning(
f"Error getting Wikipedia content for {page_title}: {e}"
)
return None
def parse_wikipedia_content(self, content: str, mep_name: str) -> Dict[str, str]:
def parse_wikipedia_content(
self, content: str, mep_name: str
) -> Dict[str, str]:
"""
Parse Wikipedia content to extract structured personality information.
Args:
content: Raw Wikipedia content
mep_name: Name of the MEP
Returns:
Dictionary of parsed personality information
"""
personality_data = {
'summary': '',
'early_life': '',
'political_career': '',
'political_views': '',
'policy_focus': '',
'achievements': '',
'controversies': '',
'personal_life': '',
'education': '',
'professional_background': '',
'party_affiliations': '',
'committee_experience': '',
'voting_record': '',
'public_statements': '',
'interests': '',
'languages': '',
'awards': '',
'publications': '',
'social_media': ''
"summary": "",
"early_life": "",
"political_career": "",
"political_views": "",
"policy_focus": "",
"achievements": "",
"controversies": "",
"personal_life": "",
"education": "",
"professional_background": "",
"party_affiliations": "",
"committee_experience": "",
"voting_record": "",
"public_statements": "",
"interests": "",
"languages": "",
"awards": "",
"publications": "",
"social_media": "",
}
# Extract summary (first paragraph)
paragraphs = content.split('\n\n')
paragraphs = content.split("\n\n")
if paragraphs:
personality_data['summary'] = paragraphs[0][:1000] # Limit summary length
personality_data["summary"] = paragraphs[0][
:1000
] # Limit summary length
# Look for specific sections
content_lower = content.lower()
# Early life and education
early_life_patterns = [
r'early life[^.]*\.',
r'born[^.]*\.',
r'childhood[^.]*\.',
r'grew up[^.]*\.',
r'education[^.]*\.'
r"early life[^.]*\.",
r"born[^.]*\.",
r"childhood[^.]*\.",
r"grew up[^.]*\.",
r"education[^.]*\.",
]
for pattern in early_life_patterns:
matches = re.findall(pattern, content_lower, re.IGNORECASE)
matches = re.findall(
pattern, content_lower, re.IGNORECASE
)
if matches:
personality_data['early_life'] = ' '.join(matches[:3]) # Take first 3 matches
personality_data["early_life"] = " ".join(
matches[:3]
) # Take first 3 matches
break
# Political career
political_patterns = [
r'political career[^.]*\.',
r'elected[^.]*\.',
r'parliament[^.]*\.',
r'minister[^.]*\.',
r'party[^.]*\.'
r"political career[^.]*\.",
r"elected[^.]*\.",
r"parliament[^.]*\.",
r"minister[^.]*\.",
r"party[^.]*\.",
]
for pattern in political_patterns:
matches = re.findall(pattern, content_lower, re.IGNORECASE)
matches = re.findall(
pattern, content_lower, re.IGNORECASE
)
if matches:
personality_data['political_career'] = ' '.join(matches[:5]) # Take first 5 matches
personality_data["political_career"] = " ".join(
matches[:5]
) # Take first 5 matches
break
# Political views
views_patterns = [
r'political views[^.]*\.',
r'positions[^.]*\.',
r'advocates[^.]*\.',
r'supports[^.]*\.',
r'opposes[^.]*\.'
r"political views[^.]*\.",
r"positions[^.]*\.",
r"advocates[^.]*\.",
r"supports[^.]*\.",
r"opposes[^.]*\.",
]
for pattern in views_patterns:
matches = re.findall(pattern, content_lower, re.IGNORECASE)
matches = re.findall(
pattern, content_lower, re.IGNORECASE
)
if matches:
personality_data['political_views'] = ' '.join(matches[:3])
personality_data["political_views"] = " ".join(
matches[:3]
)
break
# Policy focus
policy_patterns = [
r'policy[^.]*\.',
r'focus[^.]*\.',
r'issues[^.]*\.',
r'legislation[^.]*\.'
r"policy[^.]*\.",
r"focus[^.]*\.",
r"issues[^.]*\.",
r"legislation[^.]*\.",
]
for pattern in policy_patterns:
matches = re.findall(pattern, content_lower, re.IGNORECASE)
matches = re.findall(
pattern, content_lower, re.IGNORECASE
)
if matches:
personality_data['policy_focus'] = ' '.join(matches[:3])
personality_data["policy_focus"] = " ".join(
matches[:3]
)
break
# Achievements
achievement_patterns = [
r'achievements[^.]*\.',
r'accomplishments[^.]*\.',
r'success[^.]*\.',
r'won[^.]*\.',
r'received[^.]*\.'
r"achievements[^.]*\.",
r"accomplishments[^.]*\.",
r"success[^.]*\.",
r"won[^.]*\.",
r"received[^.]*\.",
]
for pattern in achievement_patterns:
matches = re.findall(pattern, content_lower, re.IGNORECASE)
matches = re.findall(
pattern, content_lower, re.IGNORECASE
)
if matches:
personality_data['achievements'] = ' '.join(matches[:3])
personality_data["achievements"] = " ".join(
matches[:3]
)
break
return personality_data
def create_personality_profile(self, mep_data: Dict[str, str]) -> MEPPersonalityProfile:
def create_personality_profile(
self, mep_data: Dict[str, str]
) -> MEPPersonalityProfile:
"""
Create a personality profile for an MEP.
Args:
mep_data: MEP data from XML file
Returns:
MEPPersonalityProfile object
"""
mep_name = mep_data['full_name']
country = mep_data['country']
mep_name = mep_data["full_name"]
country = mep_data["country"]
# Search for Wikipedia page
page_title = self.search_wikipedia_page(mep_name, country)
if page_title:
# Get Wikipedia content
wiki_content = self.get_wikipedia_content(page_title)
if wiki_content:
# Parse content
personality_data = self.parse_wikipedia_content(wiki_content['extract'], mep_name)
personality_data = self.parse_wikipedia_content(
wiki_content["extract"], mep_name
)
# Create profile
profile = MEPPersonalityProfile(
full_name=mep_name,
mep_id=mep_data['mep_id'],
wikipedia_url=wiki_content['url'],
summary=personality_data['summary'],
early_life=personality_data['early_life'],
political_career=personality_data['political_career'],
political_views=personality_data['political_views'],
policy_focus=personality_data['policy_focus'],
achievements=personality_data['achievements'],
controversies=personality_data['controversies'],
personal_life=personality_data['personal_life'],
education=personality_data['education'],
professional_background=personality_data['professional_background'],
party_affiliations=personality_data['party_affiliations'],
committee_experience=personality_data['committee_experience'],
voting_record=personality_data['voting_record'],
public_statements=personality_data['public_statements'],
interests=personality_data['interests'],
languages=personality_data['languages'],
awards=personality_data['awards'],
publications=personality_data['publications'],
social_media=personality_data['social_media'],
last_updated=time.strftime("%Y-%m-%d %H:%M:%S")
mep_id=mep_data["mep_id"],
wikipedia_url=wiki_content["url"],
summary=personality_data["summary"],
early_life=personality_data["early_life"],
political_career=personality_data[
"political_career"
],
political_views=personality_data[
"political_views"
],
policy_focus=personality_data["policy_focus"],
achievements=personality_data["achievements"],
controversies=personality_data["controversies"],
personal_life=personality_data["personal_life"],
education=personality_data["education"],
professional_background=personality_data[
"professional_background"
],
party_affiliations=personality_data[
"party_affiliations"
],
committee_experience=personality_data[
"committee_experience"
],
voting_record=personality_data["voting_record"],
public_statements=personality_data[
"public_statements"
],
interests=personality_data["interests"],
languages=personality_data["languages"],
awards=personality_data["awards"],
publications=personality_data["publications"],
social_media=personality_data["social_media"],
last_updated=time.strftime("%Y-%m-%d %H:%M:%S"),
)
if self.verbose:
logger.info(f"Created personality profile for {mep_name} from Wikipedia")
logger.info(
f"Created personality profile for {mep_name} from Wikipedia"
)
return profile
# Create minimal profile if no Wikipedia data found
profile = MEPPersonalityProfile(
full_name=mep_name,
mep_id=mep_data['mep_id'],
mep_id=mep_data["mep_id"],
summary=f"{mep_name} is a Member of the European Parliament representing {country}.",
political_career=f"Currently serving as MEP for {country}.",
political_views=f"Member of {mep_data['political_group']} and {mep_data['national_party']}.",
last_updated=time.strftime("%Y-%m-%d %H:%M:%S")
last_updated=time.strftime("%Y-%m-%d %H:%M:%S"),
)
if self.verbose:
logger.warning(f"No Wikipedia data found for {mep_name}, created minimal profile")
logger.warning(
f"No Wikipedia data found for {mep_name}, created minimal profile"
)
return profile
def save_personality_profile(self, profile: MEPPersonalityProfile) -> str:
def save_personality_profile(
self, profile: MEPPersonalityProfile
) -> str:
"""
Save personality profile to JSON file.
Args:
profile: MEPPersonalityProfile object
Returns:
Path to saved file
"""
# Create safe filename
safe_name = re.sub(r'[^\w\s-]', '', profile.full_name).strip()
safe_name = re.sub(r'[-\s]+', '_', safe_name)
safe_name = re.sub(r"[^\w\s-]", "", profile.full_name).strip()
safe_name = re.sub(r"[-\s]+", "_", safe_name)
filename = f"{safe_name}_{profile.mep_id}.json"
filepath = os.path.join(self.output_dir, filename)
# Convert to dictionary and save
profile_dict = asdict(profile)
with open(filepath, 'w', encoding='utf-8') as f:
with open(filepath, "w", encoding="utf-8") as f:
json.dump(profile_dict, f, indent=2, ensure_ascii=False)
if self.verbose:
logger.info(f"Saved personality profile: {filepath}")
return filepath
def scrape_all_mep_personalities(self, xml_file: str = "EU.xml", delay: float = 1.0) -> Dict[str, str]:
def scrape_all_mep_personalities(
self, xml_file: str = "EU.xml", delay: float = 1.0
) -> Dict[str, str]:
"""
Scrape personality data for all MEPs.
Args:
xml_file: Path to EU.xml file
delay: Delay between requests to be respectful to Wikipedia
Returns:
Dictionary mapping MEP names to their personality profile file paths
"""
meps = self.extract_mep_data_from_xml(xml_file)
profile_files = {}
if self.verbose:
logger.info(f"Starting personality scraping for {len(meps)} MEPs")
logger.info(
f"Starting personality scraping for {len(meps)} MEPs"
)
for i, mep_data in enumerate(meps, 1):
mep_name = mep_data['full_name']
mep_name = mep_data["full_name"]
if self.verbose:
logger.info(f"Processing {i}/{len(meps)}: {mep_name}")
try:
# Create personality profile
profile = self.create_personality_profile(mep_data)
# Save profile
filepath = self.save_personality_profile(profile)
profile_files[mep_name] = filepath
# Respectful delay
time.sleep(delay)
except Exception as e:
logger.error(f"Error processing {mep_name}: {e}")
continue
if self.verbose:
logger.info(f"Completed personality scraping. {len(profile_files)} profiles created.")
logger.info(
f"Completed personality scraping. {len(profile_files)} profiles created."
)
return profile_files
def load_personality_profile(self, filepath: str) -> MEPPersonalityProfile:
def load_personality_profile(
self, filepath: str
) -> MEPPersonalityProfile:
"""
Load personality profile from JSON file.
Args:
filepath: Path to personality profile JSON file
Returns:
MEPPersonalityProfile object
"""
with open(filepath, 'r', encoding='utf-8') as f:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
return MEPPersonalityProfile(**data)
def get_personality_summary(self, profile: MEPPersonalityProfile) -> str:
def get_personality_summary(
self, profile: MEPPersonalityProfile
) -> str:
"""
Generate a personality summary for use in AI agent system prompts.
Args:
profile: MEPPersonalityProfile object
Returns:
Formatted personality summary
"""
summary_parts = []
if profile.summary:
summary_parts.append(f"Background: {profile.summary}")
if profile.political_career:
summary_parts.append(f"Political Career: {profile.political_career}")
summary_parts.append(
f"Political Career: {profile.political_career}"
)
if profile.political_views:
summary_parts.append(f"Political Views: {profile.political_views}")
summary_parts.append(
f"Political Views: {profile.political_views}"
)
if profile.policy_focus:
summary_parts.append(f"Policy Focus: {profile.policy_focus}")
summary_parts.append(
f"Policy Focus: {profile.policy_focus}"
)
if profile.achievements:
summary_parts.append(f"Notable Achievements: {profile.achievements}")
summary_parts.append(
f"Notable Achievements: {profile.achievements}"
)
if profile.education:
summary_parts.append(f"Education: {profile.education}")
if profile.professional_background:
summary_parts.append(f"Professional Background: {profile.professional_background}")
summary_parts.append(
f"Professional Background: {profile.professional_background}"
)
return "\n".join(summary_parts)
def main():
"""Main function to run the Wikipedia personality scraper."""
print("🏛️ WIKIPEDIA PERSONALITY SCRAPER FOR EUROSWARM PARLIAMENT")
print("=" * 70)
# Initialize scraper
scraper = WikipediaPersonalityScraper(output_dir="mep_personalities", verbose=True)
scraper = WikipediaPersonalityScraper(
output_dir="mep_personalities", verbose=True
)
# Scrape all MEP personalities
profile_files = scraper.scrape_all_mep_personalities(delay=1.0)
print(f"\n✅ Scraping completed!")
print("\n✅ Scraping completed!")
print(f"📁 Profiles saved to: {scraper.output_dir}")
print(f"📊 Total profiles created: {len(profile_files)}")
# Show sample profile
if profile_files:
sample_name = list(profile_files.keys())[0]
sample_file = profile_files[sample_name]
sample_profile = scraper.load_personality_profile(sample_file)
print(f"\n📋 Sample Profile: {sample_name}")
print("-" * 50)
print(scraper.get_personality_summary(sample_profile))
if __name__ == "__main__":
main()
main()

@ -1,5 +1,6 @@
from swarms.structs.agent import Agent
from swarms.structs.agent_builder import AgentsBuilder
from swarms.structs.agent_rearrange import AgentRearrange, rearrange
from swarms.structs.auto_swarm_builder import AutoSwarmBuilder
from swarms.structs.base_structure import BaseStructure
from swarms.structs.base_swarm import BaseSwarm
@ -66,7 +67,6 @@ from swarms.structs.multi_agent_exec import (
run_single_agent,
)
from swarms.structs.multi_agent_router import MultiAgentRouter
from swarms.structs.agent_rearrange import AgentRearrange, rearrange
from swarms.structs.round_robin import RoundRobinSwarm
from swarms.structs.sequential_workflow import SequentialWorkflow
from swarms.structs.spreadsheet_swarm import SpreadSheetSwarm

@ -21,6 +21,13 @@ from typing import (
import toml
import yaml
from litellm import model_list
from litellm.utils import (
get_max_tokens,
supports_function_calling,
supports_parallel_function_calling,
supports_vision,
)
from loguru import logger
from pydantic import BaseModel
@ -45,7 +52,6 @@ from swarms.schemas.base_schemas import (
ChatMessageResponse,
)
from swarms.schemas.conversation_schema import ConversationSchema
from swarms.schemas.llm_agent_schema import ModelConfigOrigin
from swarms.schemas.mcp_schemas import (
MCPConnection,
)
@ -422,7 +428,6 @@ class Agent:
mcp_config: Optional[MCPConnection] = None,
top_p: Optional[float] = 0.90,
conversation_schema: Optional[ConversationSchema] = None,
aditional_llm_config: Optional[ModelConfigOrigin] = None,
llm_base_url: Optional[str] = None,
llm_api_key: Optional[str] = None,
rag_config: Optional[RAGConfig] = None,
@ -430,8 +435,8 @@ class Agent:
output_raw_json_from_tool_call: bool = False,
summarize_multiple_images: bool = False,
tool_retry_attempts: int = 3,
speed_mode: str = None,
reasoning_prompt_on: bool = True,
dynamic_context_window: bool = True,
*args,
**kwargs,
):
@ -562,7 +567,6 @@ class Agent:
self.mcp_config = mcp_config
self.top_p = top_p
self.conversation_schema = conversation_schema
self.aditional_llm_config = aditional_llm_config
self.llm_base_url = llm_base_url
self.llm_api_key = llm_api_key
self.rag_config = rag_config
@ -572,8 +576,8 @@ class Agent:
)
self.summarize_multiple_images = summarize_multiple_images
self.tool_retry_attempts = tool_retry_attempts
self.speed_mode = speed_mode
self.reasoning_prompt_on = reasoning_prompt_on
self.dynamic_context_window = dynamic_context_window
# Initialize the feedback
self.feedback = []
@ -676,17 +680,15 @@ class Agent:
# Initialize the short term memory
memory = Conversation(
name=f"{self.agent_name}_conversation",
system_prompt=prompt,
user=self.user_name,
rules=self.rules,
token_count=False,
message_id_on=False,
time_enabled=True,
)
# Add the system prompt to the conversation
memory.add(
role="system",
content=prompt,
dynamic_context_window=self.dynamic_context_window,
tokenizer_model_name=self.model_name,
context_length=self.context_length,
)
return memory
@ -888,11 +890,7 @@ class Agent:
Returns:
bool: True if model supports vision and image is provided, False otherwise.
"""
from litellm.utils import (
supports_function_calling,
supports_parallel_function_calling,
supports_vision,
)
# Only check vision support if an image is provided
if img is not None:
@ -1294,8 +1292,6 @@ class Agent:
self._handle_run_error(error)
def __handle_run_error(self, error: any):
import traceback
if self.autosave is True:
self.save()
log_agent_data(self.to_dict())
@ -1539,11 +1535,6 @@ class Agent:
raise
def reliability_check(self):
from litellm import model_list
from litellm.utils import (
get_max_tokens,
supports_function_calling,
)
if self.system_prompt is None:
logger.warning(

@ -1,21 +1,21 @@
import traceback
import concurrent.futures
import datetime
import inspect
import json
import os
import traceback
import uuid
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Literal,
Optional,
Union,
Literal,
Any,
)
import yaml
import inspect
from swarms.utils.any_to_str import any_to_str
from swarms.utils.litellm_tokenizer import count_tokens
@ -26,6 +26,18 @@ if TYPE_CHECKING:
from loguru import logger
# Define available providers
providers = Literal[
"mem0",
"in-memory",
"supabase",
"redis",
"sqlite",
"duckdb",
"pulsar",
]
def generate_conversation_id():
"""Generate a unique conversation ID."""
return str(uuid.uuid4())
@ -50,18 +62,6 @@ def get_conversation_dir():
return conversation_dir
# Define available providers
providers = Literal[
"mem0",
"in-memory",
"supabase",
"redis",
"sqlite",
"duckdb",
"pulsar",
]
def _create_backend_conversation(backend: str, **kwargs):
"""
Create a backend conversation instance based on the specified backend type.
@ -183,9 +183,9 @@ class Conversation:
name: str = "conversation-test",
system_prompt: Optional[str] = None,
time_enabled: bool = False,
autosave: bool = False, # Changed default to False
autosave: bool = False,
save_filepath: str = None,
load_filepath: str = None, # New parameter to specify which file to load from
load_filepath: str = None,
context_length: int = 8192,
rules: str = None,
custom_rules_prompt: str = None,
@ -211,6 +211,8 @@ class Conversation:
redis_data_dir: Optional[str] = None,
conversations_dir: Optional[str] = None,
export_method: str = "json",
dynamic_context_window: bool = True,
caching: bool = True,
*args,
**kwargs,
):
@ -249,6 +251,8 @@ class Conversation:
self.auto_persist = auto_persist
self.redis_data_dir = redis_data_dir
self.export_method = export_method
self.dynamic_context_window = dynamic_context_window
self.caching = caching
if self.name is None:
self.name = id
@ -933,7 +937,15 @@ class Conversation:
# Fallback to in-memory implementation
pass
elif self.dynamic_context_window is True:
return self.dynamic_auto_chunking()
else:
return self._return_history_as_string_worker()
def _return_history_as_string_worker(self):
formatted_messages = []
for message in self.conversation_history:
formatted_messages.append(
f"{message['role']}: {message['content']}"
@ -1778,20 +1790,38 @@ class Conversation:
pass
self.conversation_history = []
def dynamic_auto_chunking(self):
all_tokens = self._return_history_as_string_worker()
total_tokens = count_tokens(
all_tokens, self.tokenizer_model_name
)
if total_tokens > self.context_length:
# Get the difference between the count_tokens and the context_length
difference = total_tokens - self.context_length
# Slice the first difference number of messages and contents from the beginning of the conversation history
new_history = all_tokens[difference:]
return new_history
# # Example usage
# # conversation = Conversation()
# conversation = Conversation(token_count=True)
# Example usage
# conversation = Conversation()
# conversation = Conversation(token_count=True, context_length=14)
# conversation.add("user", "Hello, how are you?")
# conversation.add("assistant", "I am doing well, thanks.")
# conversation.add("user", "What is the weather in Tokyo?")
# print(conversation.dynamic_auto_chunking())
# # conversation.add(
# # "assistant", {"name": "tool_1", "output": "Hello, how are you?"}
# # )
# # print(conversation.return_json())
# )
# print(conversation.return_json())
# # # print(conversation.get_last_message_as_string())
# # print(conversation.get_last_message_as_string())
# print(conversation.return_json())
# # # conversation.add("assistant", "I am doing well, thanks.")
# # # # print(conversation.to_json())
# # print(type(conversation.to_dict()))
# # print(conversation.to_yaml())
# # conversation.add("assistant", "I am doing well, thanks.")
# # # print(conversation.to_json())
# print(type(conversation.to_dict()))
# print(conversation.to_yaml())

Loading…
Cancel
Save