Test Drive: Swarm
Published on .
Today I’m test driving swarm, an “Educational framework exploring ergonomic, lightweight multi-agent orchestration. Managed by OpenAI Solution team.” My task will be to get the agents to take a customer problem, define requirements, and write some code. The process is simple:
- Define your agent’s instructions
- Define any tools
- Run the swam
Adding the step for the PM to define requirements does seem to help the SWE agent. I like that it prints everything so you can see the tool calls that are being made.
from swarm import Swarm, Agent
from rich import print
client = Swarm()
def ask_for_code(requirements):
return swe
pm = Agent(
name="Product Manager",
instructions = "Based on the problem statment, you produce simple requirements for the software engineer. You ask the software engineer for code with the requirements you determine.",
functions=[ask_for_code],
)
swe = Agent(
name="Software Engineer",
instructions="You are a python Flask expert who implements solutions in python Flask. You produce code.",
)
response = client.run(
agent=pm,
messages=[{"role": "user", "content": "It takes too long to get a weather forecast, so I need you to implement a script that calls the NWS weather API"}],
model_override='gpt-4o-mini',
)
Results
Certainly! Below is a Python Flask application that calls the NWS (National Weather Service) weather API to get the current weather forecast based on user-provided latitude and longitude. The application handles errors gracefully and returns the current temperature and weather conditions.
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
NWS_API_BASE_URL = "https://api.weather.gov/points"
def get_weather_forecast(latitude, longitude):
# Getting the forecast URL from NWS API using latitude and longitude
forecast_url = f"{NWS_API_BASE_URL}/{latitude},{longitude}"
try:
response = requests.get(forecast_url)
response.raise_for_status() # Raise an error for bad responses
forecast_data = response.json()
# Extract the forecast URL from the response
forecast_api_url = forecast_data['properties']['forecast']
# Fetching the actual forecast
forecast_response = requests.get(forecast_api_url)
forecast_response.raise_for_status()
forecast_details = forecast_response.json()
# Extracting current conditions
current_conditions = forecast_details['properties']['currentObservations'][0]
temperature = current_conditions['temperature']
weather = current_conditions['shortForecast']
return {
'temperature': temperature,
'weather': weather
}
except requests.exceptions.HTTPError as http_err:
return {'error': f'HTTP error occurred: {http_err}'}, 400
except Exception as err:
return {'error': f'An error occurred: {err}'}, 400
@app.route('/weather', methods=['GET'])
def weather():
# Getting latitude and longitude from query parameters
latitude = request.args.get('lat')
longitude = request.args.get('lon')
if not latitude or not longitude:
return jsonify({'error': 'Please provide latitude and longitude as query parameters.'}), 400
weather_data = get_weather_forecast(latitude, longitude)
return jsonify(weather_data)
if __name__ == '__main__':
app.run(debug=True)
Instructions to Run:
Install Flask and Requests: Make sure you have
theFlask
andrequests
libraries installed.
You can install them using pip:pip install Flask requests
Save the code: Save the above code in a file
namedapp.py
.Run the Flask app:
python app.py
Accessing the API: You can access the weather
forecast by going to the following endpoint in your web browser or using
a tool like Postman:http://127.0.0.1:5000/weather?lat=38.4247341&lon=-86.9624086
Make sure to replace
lat
andlon
with the
desired latitude and longitude values.
This application will return the current temperature and weather conditions for the specified location.