OpenAI Swarm

[tcdent]

This addresses #244.

TODO

• restore agentops tool wrapper functionality

[codecov]

Codecov Report

Attention: Patch coverage is 84.06593% with 29 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
agentstack/frameworks/openai_swarm.py	83.67%	19 Missing and 5 partials ⚠️
agentstack/frameworks/__init__.py	82.14%	5 Missing ⚠️

📢 Thoughts on this report? Let us know!

[tcdent]

helllo_alex gets implemented like below.

Tasks get placed before Agents and prefix the context.

There's no real way to define delegation; if you add an agent as a tool of another agent, it will delegate, but then what? the _handoff helper let's you pass an agent by name as a tool.

_get_first_task gives the app a starting point, but is less than ideal.

Limitations of the framework more than anything. Read the source, it's like 200 lines.

from swarm import Swarm, Agent
import agentstack


class FoobarStack:

    def _handoff(self, agent_name: str):
        """Tool for handing off to another agent."""
        agent = getattr(self, agent_name)
        def func(context_variables: list[str]):
            return agent(context_variables)
        return func

    def _get_first_task(self):
        """Get the first task."""
        task_name = agentstack.get_all_task_names()[0]
        return getattr(self, task_name)()

    @agentstack.agent
    def alex(self, messages: list[str] = []):
        agent_config = agentstack.get_agent('alex')
        return Agent(
            name=agent_config.name, 
            model=agent_config.model, 
            instructions="\n".join([agent_config.prompt, *messages, ]),
            functions=[*agentstack.tools['file_read'], ],
        )

    @agentstack.task
    def hello_world(self):
        task_config = agentstack.get_task('hello_world')
        messages = [
            task_config.prompt, 
        ]
        agent = getattr(self, task_config.agent)
        return agent(messages)

    def run(self, inputs: list[str]):
        app = Swarm()
        response = app.run(
            agent=self._get_first_task(), 
            messages=[], 
            context_variables=inputs, 
            debug=agentstack.conf.DEBUG, 
        )

        for message in response.messages:
            agentstack.log.info(message['content'])

[bboynton97]

➜ agentstack run --debug
Running your agent...
An error occurred: 
Failed to import AgentStack project at: /Users/braelynboynton/Developer/agentstack/AgentStack/examples/swarm_test
cannot import name 'SwarmtestSwarm' from 'stack' (/Users/braelynboynton/Developer/agentstack/AgentStack/examples/swarm_test/src/stack.py)
DEBUG: Full traceback:
Traceback (most recent call last):
  File "/Users/braelynboynton/Developer/agentstack/AgentStack/agentstack/cli/run.py", line 123, in run_project
    project_main = _import_project_module(conf.PATH)
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/braelynboynton/Developer/agentstack/AgentStack/agentstack/cli/run.py", line 92, in _import_project_module
    spec.loader.exec_module(project_module)
  File "<frozen importlib._bootstrap_external>", line 999, in exec_module
  File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
  File "/Users/braelynboynton/Developer/agentstack/AgentStack/examples/swarm_test/src/main.py", line 4, in <module>
    from stack import SwarmtestSwarm
ImportError: cannot import name 'SwarmtestSwarm' from 'stack' (/Users/braelynboynton/Developer/agentstack/AgentStack/examples/swarm_test/src/stack.py). Did you mean: 'SwarmtestStack'?

i think just a typo, i can fix

[bboynton97]

it doesnt look like we're automatically ending the agentops session after the swarm finishes

we might look for callback functionality for async. For sync, just run agentops.end_session(status="Success") after .run()

[bboynton97]

incredible. iterating through tasks was smart. waiting for tests to pass then good to merge