from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool import datetime import requests import pytz import yaml from typing import List, Dict, Optional import tkapi from tkapi.fractie import Fractie, FractieFilter from tkapi.persoon import PersoonFilter, PersoonNevenfunctie from tkapi.util import queries from tkapi.zaak import Zaak, ZaakSoort from tools.final_answer import FinalAnswerTool from Gradio_UI import GradioUI api = tkapi.TKApi() @tool def get_kamervraag_context(zaak_nummer: str) -> Dict[str, any]: """ Retrieves detailed context information for a specific parliamentary item by its number. Args: zaak_nummer: The unique identifier of the parliamentary item (e.g., '2024Z00541') Returns: Dict[str, any]: Dictionary containing detailed item information: - 'nummer': Unique identifier of the item - 'soort': Type of the item - 'onderwerp': Subject/topic of the item - 'datum': Date the item was started - 'afgedaan': Whether the item has been handled/answered - 'indieners': List of submitters with their party - 'documenten': List of related documents with their details - 'besluiten': List of related decisions - 'activiteiten': List of related activities Example return value: { 'nummer': '2024Z00541', 'soort': 'Motie', 'onderwerp': 'Motie van de leden over uitingen van de schutterijcultuur', 'datum': datetime.date(2024, 1, 18), 'afgedaan': True, 'indieners': ['Geert Wilders (PVV)', 'Martin Bosma (PVV)'], 'documenten': [ { 'nummer': '36410-VIII-56', 'soort': 'Motie', 'titel': 'Motie van de leden...', 'datum': datetime.date(2024, 1, 18) } ], 'besluiten': ['Aangenomen op 23 januari 2024'], 'activiteiten': [ { 'soort': 'Stemmingen', 'datum': datetime.date(2024, 1, 23), 'status': 'Afgerond' } ] } """ try: # Initialize API api = tkapi.TKApi(verbose=False) # Create filter for specific zaak filter = Zaak.create_filter() filter.filter_nummer(zaak_nummer) # Get the zaak zaken = api.get_zaken(filter=filter) zaak = next(iter(zaken), None) if not zaak: print(f"No parliamentary item found with number {zaak_nummer}") return {} # Collect basic information result = { 'nummer': zaak.nummer, 'soort': zaak.soort.value if zaak.soort else None, 'onderwerp': zaak.onderwerp, 'datum': zaak.gestart_op, 'afgedaan': zaak.afgedaan, 'indieners': [], 'documenten': [], 'besluiten': [], 'activiteiten': [] } # Get indieners (actors) actors = zaak.actors if actors: for actor in actors: if hasattr(actor, 'persoon') and actor.persoon: persoon = actor.persoon fractie_str = "" if hasattr(persoon, 'fractie') and persoon.fractie: fractie_str = f" ({persoon.fractie.afkorting})" result['indieners'].append( f"{persoon.roepnaam} {persoon.achternaam}{fractie_str}" ) #a # Get related documents for doc in zaak.documenten: doc_info = { 'nummer': doc.nummer if hasattr(doc, 'nummer') else None, 'soort': doc.soort.value if hasattr(doc, 'soort') and doc.soort else None, 'titel': doc.titel if hasattr(doc, 'titel') else None, 'datum': doc.datum if hasattr(doc, 'datum') else None } result['documenten'].append(doc_info) # Get decisions for besluit in zaak.besluiten: if hasattr(besluit, 'tekst') and besluit.tekst: result['besluiten'].append(besluit.tekst) # Get activities for activiteit in zaak.activiteiten: act_info = { 'soort': activiteit.soort.value if hasattr(activiteit, 'soort') and activiteit.soort else None, 'datum': activiteit.datum if hasattr(activiteit, 'datum') else None, 'status': activiteit.status if hasattr(activiteit, 'status') else None } result['activiteiten'].append(act_info) return result except Exception as e: print(f"Error fetching context for parliamentary item: {str(e)}") return {} @tool def get_kamervragen( start_datetime: datetime.datetime, end_datetime: datetime.datetime, only_answered: Optional[bool] = None, zaak_soort: ZaakSoort = ZaakSoort.SCHRIFTELIJKE_VRAGEN ) -> List[Dict[str, any]]: """ Retrieves basic information about parliamentary items (kamervragen, moties, etc.) within a specified date range. Args: start_datetime: Start date as datetime object end_datetime: End date as datetime object only_answered: Optional filter for answered items (True = only answered, False = only unanswered, None = both) zaak_soort: Type of parliamentary item to filter on. Options from ZaakSoort enum: - AMENDEMENT: Amendments - BRIEF_COMMISSIE: Committee letters - BRIEF_REGERING: Government letters - BRIEF_LID: Member letters - INITIATIEF_NOTA: Initiative notes - INITIATIEF_WETGEVING: Initiative legislation - INTERPELLATIE: Interpellations - LIJST_MET_VRAGEN: List of questions - MEDEDELINGEN: Announcements - MONDELINGE_VRAGEN: Oral questions - MOTIE: Motions - OVERIG: Other - PARLEMENTAIR_DOCUMENT: Parliamentary documents - RAPPORT: Reports - SCHRIFTELIJKE_VRAGEN: Written questions (default) - VERSLAG_COMMISSIE: Committee reports - VOORDRACHT: Nominations - WETSVOORSTEL: Bill proposals Returns: List[Dict[str, any]]: List of dictionaries containing basic item information: - 'nummer': Unique identifier of the item - 'soort': Type of the item - 'onderwerp': Subject/topic of the item - 'datum': Date the item was started (as datetime.date) - 'afgedaan': Boolean indicating if the item has been handled/answered Example return value: [ { 'nummer': '2024Z00541', 'soort': 'Motie', 'onderwerp': 'Motie van de leden over uitingen van de schutterijcultuur', 'datum': datetime.date(2024, 1, 18), 'afgedaan': True } ] """ try: # Initialize API api = TKApi(verbose=False) # Create filter filter = Zaak.create_filter() filter.filter_soort(zaak_soort) filter.filter_date_range(start_datetime, end_datetime) if only_answered is not None: filter.filter_afgedaan(only_answered) # Get items zaken = api.get_zaken(filter=filter) # Format results results = [] for zaak in zaken: zaak_info = { 'nummer': zaak.nummer, 'soort': zaak.soort.value if zaak.soort else None, 'onderwerp': zaak.onderwerp, 'datum': zaak.gestart_op, 'afgedaan': zaak.afgedaan } results.append(zaak_info) return results except Exception as e: print(f"Error fetching parliamentary items: {str(e)}") return [] @tool def get_kamerlid_reizen(achternaam: str) -> List[Dict[str, str]]: """ Retrieves all registered travels for a specific parliament member by their last name. Args: achternaam: The last name of the parliament member to search for (case-sensitive, partial matches are supported) Returns: List[Dict[str, str]]: A list of dictionaries containing travel information: - 'bestemming': Destination of the travel - 'doel': Purpose of the travel - 'van': Start date of the travel - 'tot': End date of the travel - 'betaald_door': Who paid for the travel Example return value: [ { 'bestemming': 'Brussels', 'doel': 'EU Parliament meeting', 'van': '2024-01-15', 'tot': '2024-01-17', 'betaald_door': 'European Parliament' } ] Note: If multiple parliament members match the given last name, travels for all matching members will be returned. """ try: # Initialize API api = tkapi.TKApi(verbose=False) # Create person filter filter = PersoonFilter() filter.filter_achternaam(achternaam) # Get matching persons persons = api.get_personen(filter=filter) # Collect all travels all_travels = [] for person in persons: for reis in person.reizen: travel_info = { 'bestemming': reis.bestemming, 'doel': reis.doel, 'van': str(reis.van) if reis.van else None, 'tot': str(reis.tot_en_met) if reis.tot_en_met else None, 'betaald_door': reis.betaald_door } all_travels.append(travel_info) return all_travels except Exception as e: print(f"Error fetching travel data: {str(e)}") return [] @tool def Get_Active_Fractions() -> list[dict[str, str | int]]: """ Retrieves a list of all currently active fractions (political parties) in the Dutch parliament. Returns: List[Dict[str, Union[str, int]]]: A list of dictionaries containing information about each fraction: - 'naam': Full name of the political party (string) - 'afkorting': Abbreviation/acronym of the party (string) - 'zetels_aantal': Number of seats the party holds (integer) Example return value: [ { 'naam': 'Volkspartij voor Vrijheid en Democratie', 'afkorting': 'VVD', 'zetels_aantal': 34 }, { 'naam': 'Democraten 66', 'afkorting': 'D66', 'zetels_aantal': 24 } ] """ try: # Initialize API client api = tkapi.TKApi(verbose=False) # Create filter for active fractions filter = Fractie.create_filter() filter.filter_actief() # Get all active fractions active_fractions = api.get_fracties(filter=filter) # Format the results results = [] for fraction in active_fractions: fraction_info = { 'naam': fraction.naam, 'afkorting': fraction.afkorting, 'zetels_aantal': fraction.zetels_aantal } results.append(fraction_info) return results except Exception as e: print(f"Error fetching fraction data: {str(e)}") return [] @tool def get_fraction_members(fractie_naam: str) -> List[Dict[str, str]]: """ Retrieves all current members of a specific fraction (political party) in the Dutch parliament. Args: fractie_naam: The name of the fraction/party to search for (case-sensitive). Can be either the full name (e.g. 'Volkspartij voor Vrijheid en Democratie') or the abbreviation (e.g. 'VVD') Returns: List[Dict[str, str]]: A list of dictionaries containing member information: - 'achternaam': Last name of the member - 'geboortedatum': Date of birth - 'van': Start date of membership - 'fractie': Name of the fraction - 'fractie_afkorting': Abbreviation of the fraction Example return value: [ { 'achternaam': 'van der Berg', 'geboortedatum': '1975-03-15', 'van': '2023-12-06', 'fractie': 'Volkspartij voor Vrijheid en Democratie', 'fractie_afkorting': 'VVD' } ] Note: - This only returns currently active members - The search works for both full names and abbreviations - If no fraction is found with the given name, an empty list is returned """ try: # Initialize API api = tkapi.TKApi(verbose=False) # First try to find the fraction filter = FractieFilter() filter.filter_actief() # Only get active fractions fracties = api.get_fracties(filter=filter) # Find matching fraction (check both full name and abbreviation) target_fractie = None for fractie in fracties: if fractie_naam.upper() in [fractie.naam.upper(), fractie.afkorting.upper()]: target_fractie = fractie break if not target_fractie: print(f"No active fraction found with name or abbreviation: {fractie_naam}") return [] # Get active members for the fraction members = target_fractie.leden_actief # Format member information member_list = [] for lid in members: persoon = lid.persoon member_info = { 'achternaam': persoon.achternaam, 'geboortedatum': str(persoon.geboortedatum) if persoon.geboortedatum else None, 'van': str(lid.van) if lid.van else None, 'fractie': target_fractie.naam, 'fractie_afkorting': target_fractie.afkorting } member_list.append(member_info) return member_list except Exception as e: print(f"Error fetching fraction members: {str(e)}") return [] @tool def get_kamerlid_geschenken(achternaam: str) -> List[Dict[str, str]]: """ Retrieves all registered gifts for a specific parliament member by their last name. Args: achternaam: The last name of the parliament member to search for (case-sensitive, partial matches are supported) Returns: List[Dict[str, str]]: A list of dictionaries containing gift information: - 'persoon': Name of the parliament member - 'omschrijving': Description of the gift - 'datum': Date the gift was received - 'waarde': Value of the gift if specified Example return value: [ { 'persoon': 'John van der Berg', 'omschrijving': 'Book "Politics in the Netherlands"', 'datum': '2024-01-15', 'waarde': '€45' } ] Note: If multiple parliament members match the given last name, gifts for all matching members will be returned. """ try: # Initialize API api = tkapi.TKApi(verbose=False) # Create person filter filter = PersoonFilter() filter.filter_achternaam(achternaam) # Get matching persons persons = api.get_personen(filter=filter) # Collect all gifts all_gifts = [] for person in persons: for geschenk in person.geschenken: gift_info = { 'persoon': f"{person.roepnaam} {person.achternaam}".strip(), 'omschrijving': geschenk.omschrijving, 'datum': str(geschenk.datum) if geschenk.datum else None, 'waarde': geschenk.waarde if hasattr(geschenk, 'waarde') else None } all_gifts.append(gift_info) return all_gifts except Exception as e: print(f"Error fetching gift data: {str(e)}") return [] @tool def get_kamerlid_nevenfuncties(achternaam: str) -> List[Dict[str, str]]: """ Retrieves all registered secondary functions (nevenfuncties) for a specific parliament member by their last name. Args: achternaam: The last name of the parliament member to search for (case-sensitive, partial matches are supported) Returns: List[Dict[str, str]]: A list of dictionaries containing secondary function information: - 'persoon': Name of the parliament member - 'omschrijving': Description of the function - 'vergoeding': Compensation details if any - 'van': Start date of the function - 'tot': End date of the function (if applicable) - 'periode': Period specification (if available) Example return value: [ { 'persoon': 'John van der Berg', 'omschrijving': 'Board member of Foundation XYZ', 'vergoeding': 'Unpaid', 'van': '2023-01-01', 'tot': None, 'periode': 'Current' } ] Note: If multiple parliament members match the given last name, secondary functions for all matching members will be returned. """ try: # Initialize API api = tkapi.TKApi(verbose=False) # Create person filter filter = PersoonFilter() filter.filter_achternaam(achternaam) # Get matching persons persons = api.get_personen(filter=filter) # Collect all secondary functions all_functions = [] for person in persons: for functie in person.nevenfuncties: function_info = { 'persoon': f"{person.roepnaam} {person.achternaam}".strip(), 'omschrijving': functie.omschrijving, 'vergoeding': functie.vergoeding if hasattr(functie, 'vergoeding') else None, 'van': str(functie.van) if functie.van else None, 'tot': str(functie.tot) if hasattr(functie, 'tot') and functie.tot else None, 'periode': functie.periode if hasattr(functie, 'periode') else None } all_functions.append(function_info) return all_functions except Exception as e: print(f"Error fetching secondary function data: {str(e)}") return [] @tool def get_current_time_in_timezone(timezone: str) -> str: """A tool that fetches the current local time in a specified timezone. Args: timezone: A string representing a valid timezone (e.g., 'America/New_York'). """ try: # Create timezone object tz = pytz.timezone(timezone) # Get current time in that timezone local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") return f"The current local time in {timezone} is: {local_time}" except Exception as e: return f"Error fetching time for timezone '{timezone}': {str(e)}" final_answer = FinalAnswerTool() model = HfApiModel( max_tokens=2096, temperature=0.5, model_id='Qwen/Qwen2.5-Coder-32B-Instruct', custom_role_conversions=None, ) # Import tool from Hub image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True) with open("prompts.yaml", 'r') as stream: prompt_templates = yaml.safe_load(stream) agent = CodeAgent( model=model, tools=[final_answer, Get_Active_Fractions, get_fraction_members, get_kamerlid_reizen, get_kamerlid_geschenken, get_kamerlid_nevenfuncties, get_kamervragen, get_kamervraag_context, get_current_time_in_timezone], ## the given tools max_steps=6, verbosity_level=1, grammar=None, planning_interval=None, name=None, description=None, prompt_templates=prompt_templates ) GradioUI(agent).launch()