SwarmZero SDK

This library provides you with an easy way to create and run AI Agents and Swarms of Agents.

Supported LLM Providers:

• OpenAI
• Azure OpenAI
• Anthropic
• MistralAI
• Gemini
• Nebius
• Ollama
• AWS Bedrock
• OpenRouter

Project Requirements

• Python >= 3.11

Installation

You can either directly install with pip:

pip install swarmzero

Or can either directly install with poetry:

poetry add swarmzero

Or add it to your requirements.txt file:

...
swarmzero==x.y.z
...

Environment Setup

You need to specify an OPENAI_API_KEY in a .env file in this directory.

Make a copy of the .env.example file and rename it to .env.

Configuration Setup

To use a configuration file with your Agent, follow these steps:

1. Create a Configuration File:

   • Create a TOML or YAML file (e.g., swarmzero_config.toml or swarmzero_config.yaml) in your project directory. (See swarmzero_config_example.toml or swarmzero_config_example.yaml).

2. Create an SDK Context:

   • Create an instance of SDKContext with the path to your configuration file.
   • The SDKContext allows you to manage configurations, resources, and utilities across your SwarmZero Agents more efficiently.

   from swarmzero.sdk_context import SDKContext
   
   sdk_context = SDKContext(config_path="./swarmzero_config.toml")
   # or use a YAML file
   # sdk_context = SDKContext(config_path="./swarmzero_config.yaml")
   

3. Specify the Configuration Path:

   • When creating a Agent instance, provide the relative or absolute path to your configuration file.
   • Agent will use the configuration from the SDK Context. If you have one agent you can directly pass the config_path it will create the sdk_context for you.

   from swarmzero import Agent
   
   simple_agent = Agent(
       name="Simple Agent",
       functions=[],
       instruction="your instructions for this agent's goal",
       # sdk_context=sdk_context
       config_path="./swarmzero_config.toml" 
       # or use a YAML file
       # config_path="./swarmzero_config.yaml" 
   )
   

Usage

More detailed examples can be found at https://github.com/swarmzero/examples

First import the Agent class:

from swarmzero import Agent

Load your environment variables:

from dotenv import load_dotenv
load_dotenv()

Then create a Agent instance:

my_agent = Agent(
    name="my_agent",
    functions=[],
    instruction="your instructions for this agent's goal",
)

Then, run your agent:

my_agent.run()

Finally, call the API endpoint, /api/v1/chat, to see the result:

curl --request POST \
  --url http://localhost:8000/api/v1/chat \
  --header 'Content-Type: multipart/form-data' \
  --form 'user_id="test"' \
  --form 'session_id="test"' \
  --form 'chat_data={ "messages": [ { "role": "user", "content": "Who is Satoshi Nakamoto?" } ] }'

Adding tools

You can create tools that help your agent handle more complex tasks. Here's an example:

import os
from typing import Optional, Dict
from web3 import Web3
from swarmzero import Agent
from dotenv import load_dotenv

load_dotenv()

rpc_url = os.getenv("RPC_URL") # add an ETH Mainnet HTTP RPC URL to your `.env` file

def get_transaction_receipt(transaction_hash: str) -> Optional[Dict]:
    """
    Fetches the receipt of a specified transaction on the Ethereum blockchain and returns it as a dictionary.

    :param transaction_hash: The hash of the transaction to fetch the receipt for.
    :return: A dictionary containing the transaction receipt details, or None if the transaction cannot be found.
    """
    web3 = Web3(Web3.HTTPProvider(rpc_url))

    if not web3.is_connected():
        print("unable to connect to Ethereum")
        return None

    try:
        transaction_receipt = web3.eth.get_transaction_receipt(transaction_hash)
        return dict(transaction_receipt)
    except Exception as e:
        print(f"an error occurred: {e}")
        return None

if __name__ == "__main__":
    my_agent = Agent(
        name="my_agent",
        functions=[get_transaction_receipt]
    )

    my_agent.run()

    """
    [1] send a request:

    ```
    curl --request POST \
    --url http://localhost:8000/api/v1/chat \
    --header 'Content-Type: multipart/form-data' \
    --form 'user_id="test"' \
    --form 'session_id="test"' \
    --form 'chat_data={ "messages": [ { "role": "user", "content": "Who is the sender of this transaction - 0x5c504ed432cb51138bcf09aa5e8a410dd4a1e204ef84bfed1be16dfba1b22060" } ] }'
    ```

    [2] result:

    The address that initiated the transaction with hash 0x5c504ed432cb51138bcf09aa5e8a410dd4a1e204ef84bfed1be16dfba1b22060 is 0xA1E4380A3B1f749673E270229993eE55F35663b4.
    """

Creating a Swarm

You can create a swarm of agents to collaborate on complex tasks. Here's an example of how to set up and use a swarm:

from swarmzero.swarm import Swarm
from swarmzero.agent import Agent
from swarmzero.sdk_context import SDKContext

import asyncio

# Create SDK Context
sdk_context = SDKContext(config_path="./swarmzero_config_example.toml")
# or
# sdk_context = SDKContext(config_path="./swarmzero_config_example.yaml")


def save_report():
    return "save_item_to_csv"


def search_on_web():
    return "search_on_web"


# Create individual agents
agent1 = Agent(name="Research Agent", instruction="Conduct research on given topics", sdk_context=sdk_context,
               functions=[search_on_web])
agent2 = Agent(name="Analysis Agent", instruction="Analyze data and provide insights", sdk_context=sdk_context,
               functions=[save_report])
agent3 = Agent(name="Report Agent", instruction="Compile findings into a report", sdk_context=sdk_context, functions=[])

# Create swarm
swarm = Swarm(name="Research Team", description="A swarm of agents that collaborate on research tasks",
              instruction="Be helpful and collaborative", functions=[], agents=[agent1, agent2, agent3])


async def chat_with_swarm():
    return await swarm.chat("Can you analyze the following data: [1, 2, 3, 4, 5]")


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

Workflow

You can orchestrate agents, swarms, and tools in a flexible workflow. Each step can run sequentially, in parallel, conditionally, or in a loop.

from swarmzero import Workflow, WorkflowStep, StepMode
from swarmzero.sdk_context import SDKContext
import asyncio

# Create SDK Context
sdk_context = SDKContext(config_path="./swarmzero_config.toml")

# agent1, agent2, agent3, and agent4 are pre-defined Agent instances

workflow = Workflow(
    name="Research Workflow",
    description="Research and Analysis Pipeline",
    instruction="Demo workflow",
    sdk_context=sdk_context,
    steps=[
        # Sequential step - runs agent1
        WorkflowStep(runner=agent1.chat),
        # Parallel step - runs agent2 and agent3
        WorkflowStep(runner=[agent2.chat, agent3.chat], mode=StepMode.PARALLEL),
        # Loop step - repeats until condition is met
        WorkflowStep(
            runner=agent4.chat,
            mode=StepMode.LOOP,
            condition=lambda res: "done" in res,
            max_iterations=5,
        ),
    ],
)

async def run_workflow():
    return await workflow.run("Start research")

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

Nested Workflows

Workflow steps can themselves be workflows. This allows complex pipelines to be composed from smaller, reusable ones.

inner = Workflow(
    name="Inner",
    steps=[WorkflowStep(runner=agent1.chat)],
)

outer = Workflow(
    name="Outer",
    steps=[WorkflowStep(runner=inner)],
)

async def run_nested():
    return await outer.run("start")

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

Adding Retriever

You can add retriever tools to create vector embeddings and retrieve semantic information. It will create vector index for every pdf documents under 'swarmzero-data/files/user' folder and can filter files with required_exts parameter.

• SwarmZero agent supports ".md", '.mdx' ,".txt", '.csv', '.docx', '.pdf' file types.
• SwarmZero agent supports 4 type of retriever (basic, chroma, pinecone-serverless, pinecone-pod) and controlled with retrieval_tool parameter.

from swarmzero import Agent
fro