Interactive Agent

import asyncio

from dotenv import load_dotenv

import synalinks

# Activate logging for monitoring interactions
synalinks.enable_logging()

MAX_ITERATIONS = 5


# Define the calculation tool
@synalinks.utils.register_synalinks_serializable()
async def calculate(expression: str):
    """Perform calculations based on a mathematical expression.

    Args:
        expression (str): The mathematical expression to calculate, such as
            '2 + 2'. The expression can contain numbers, operators (+, -, *, /),
            parentheses, and spaces.
    """
    # Check for valid characters in the expression
    if not all(char in "0123456789+-*/(). " for char in expression):
        return {
            "result": None,
            "log": (
                "Invalid characters detected in the expression. "
                "Only numbers, operators (+, -, *, /), parentheses, and spaces "
                "are allowed."
            ),
        }
    try:
        # Safely evaluate the mathematical expression
        result = round(float(eval(expression, {"__builtins__": None}, {})), 2)
        return {
            "result": result,
            "log": "Calculation successful",
        }
    except Exception as e:
        return {
            "result": None,
            "log": f"Calculation error: {e}",
        }


async def main():
    load_dotenv()

    # Enable observability for tracing
    synalinks.enable_observability(
        tracking_uri="http://localhost:5000",
        experiment_name="interactive_math_agent",
    )

    # Initialize the language model
    language_model = synalinks.LanguageModel(
        model="gemini/gemini-3.1-flash-lite-preview",
    )

    # Define the tools available to the agent
    tools = [
        synalinks.Tool(calculate),
    ]

    # ==========================================================================
    # Create the interactive agent
    # ==========================================================================
    print("Creating the interactive agent...")

    # Set up the input structure using ChatMessages
    inputs = synalinks.Input(data_model=synalinks.ChatMessages)

    # Create the interactive agent
    # In non-autonomous mode, the agent returns ChatMessages containing both
    # tool results and assistant response
    outputs = await synalinks.FunctionCallingAgent(
        tools=tools,
        language_model=language_model,
        return_inputs_with_trajectory=True,
        autonomous=False,
    )(inputs)

    # Define the agent program
    agent = synalinks.Program(
        inputs=inputs,
        outputs=outputs,
        name="interactive_math_agent",
        description="An agent designed to handle mathematical queries interactively",
    )

    synalinks.utils.plot_program(
        agent,
        to_folder="examples",
        show_module_names=True,
        show_trainable=True,
        show_schemas=True,
    )

    # ==========================================================================
    # Run the interactive conversation
    # ==========================================================================
    print("Running the interactive agent...")

    # Initialize the conversation with a user query
    input_messages = synalinks.ChatMessages(
        messages=[
            synalinks.ChatMessage(
                role="user",
                content="Calculate the sum of 152648 and 485.",
            )
        ]
    )

    # Process the conversation through multiple iterations
    for iteration in range(MAX_ITERATIONS):
        print(f"\n--- Iteration {iteration + 1} ---")
        response = await agent(input_messages)

        # The response contains all messages including tool results
        # The last message is the assistant response
        assistant_message = response.get("messages")[-1]

        # Check if the agent wants to call tools
        tool_calls = assistant_message.get("tool_calls")
        if not tool_calls:
            # No tool calls - the agent is done and has a final response
            print("\nAgent final response:")
            print(f"  {assistant_message.get('content')}")
            print("\nConversation complete.")
            break

        # =======================================================================
        # Human-in-the-loop validation step
        # In a real application, you would present these tool calls to the user
        # via a UI or CLI and let them approve, modify, or reject them.
        # =======================================================================
        print("\nAgent wants to call the following tools:")
        for i, tool_call in enumerate(tool_calls):
            tool_name = tool_call.get("name")
            tool_args = tool_call.get("arguments")
            print(f"  [{i + 1}] {tool_name}({tool_args})")

        # Simulate user validation (in a real app, this would be interactive)
        print("\n[Simulated] User validates the tool calls...")
        print("[Simulated] Proceeding with execution...")

        # After validation, append the assistant message with tool_calls
        # The agent will execute the tools and continue in the next iteration
        input_messages.messages.append(assistant_message)


if __name__ == "__main__":
    asyncio.run(main())