Kye
8 months ago
parent
768450acd1
commit
c78d63f18d
@ -0,0 +1,6 @@
|
||||
{
|
||||
"agent_id": "<function agent_id at 0x109536fc0>",
|
||||
"agent_name": "Space Traffic Controller Agent",
|
||||
"agent_description": null,
|
||||
"LLM": "{'model_name': None, 'max_tokens': 4000, 'temperature': 0.8, 'top_k': None, 'top_p': None, 'system_prompt': 'You are a helpful assistant.', 'beam_width': None, 'num_return_sequences': None, 'seed': None, 'frequency_penalty': None, 'presence_penalty': None, 'stop_token': None, 'length_penalty': None, 'role': None, 'max_length': None, 'do_sample': None, 'early_stopping': None, 'num_beams': None, 'repition_penalty': None, 'pad_token_id': None, 'eos_token_id': None, 'bos_token_id': None, 'device': None, 'history': [], 'start_time': None, 'end_time': None, 'memory': {'input': [], 'output': [], 'task': [], 'time': [], 'role': [], 'model': []}, 'model': 'meta-llama/Meta-Llama-3-8B-Instruct'}",
|
||||
"system_prompt":
|
||||
@ -0,0 +1,6 @@
|
||||
{
|
||||
"agent_id": "<function agent_id at 0x109536fc0>",
|
||||
"agent_name": "Weather Analyst Agent",
|
||||
"agent_description": null,
|
||||
"LLM": "{'model_name': None, 'max_tokens': 4000, 'temperature': 0.8, 'top_k': None, 'top_p': None, 'system_prompt': 'You are a helpful assistant.', 'beam_width': None, 'num_return_sequences': None, 'seed': None, 'frequency_penalty': None, 'presence_penalty': None, 'stop_token': None, 'length_penalty': None, 'role': None, 'max_length': None, 'do_sample': None, 'early_stopping': None, 'num_beams': None, 'repition_penalty': None, 'pad_token_id': None, 'eos_token_id': None, 'bos_token_id': None, 'device': None, 'history': [], 'start_time': None, 'end_time': None, 'memory': {'input': [], 'output': [], 'task': [], 'time': [], 'role': [], 'model': []}, 'model': 'meta-llama/Meta-Llama-3-8B-Instruct'}",
|
||||
"system_prompt":
|
||||
@ -0,0 +1,118 @@
|
||||
# Documentation Outline for `BaseTool` Class
|
||||
|
||||
1. **Module Overview**
|
||||
2. **Installation and Setup**
|
||||
3. **Class Definition**
|
||||
4. **Attributes and Methods**
|
||||
5. **Functionality and Usage**
|
||||
- Basic Usage Examples
|
||||
- Advanced Use Cases
|
||||
6. **Common Issues and Troubleshooting**
|
||||
7. **References and Additional Resources**
|
||||
|
||||
## 1. Module Overview
|
||||
|
||||
The `BaseTool` class is a part of the `swarms` package and serves as a foundational class for creating and managing tools that can be executed with different inputs and configurations. It leverages Pydantic for input validation and includes extensive logging for easy debugging and monitoring.
|
||||
|
||||
## 2. Installation and Setup
|
||||
|
||||
To use the `BaseTool` class, ensure that you have the required dependencies installed:
|
||||
|
||||
```bash
|
||||
pip install pydantic loguru
|
||||
```
|
||||
|
||||
Include the necessary imports in your Python script:
|
||||
|
||||
```python
|
||||
from swarms.tools.base_tool import BaseTool
|
||||
```
|
||||
|
||||
## 3. Class Definition
|
||||
|
||||
`BaseTool` is designed using Pydantic's `BaseModel` to leverage type annotations and validations:
|
||||
|
||||
```python
|
||||
from pydantic import BaseModel
|
||||
|
||||
class BaseTool(BaseModel):
|
||||
# Attributes and method definitions follow
|
||||
```
|
||||
|
||||
## 4. Attributes and Methods
|
||||
|
||||
### Attributes
|
||||
|
||||
| Attribute | Type | Description |
|
||||
|---------------------|----------------------------|--------------------------------------------------------------|
|
||||
| `verbose` | `bool` | Enables verbose output, providing detailed logs. |
|
||||
| `functions` | `List[Callable[..., Any]]` | Stores a list of functions that can be managed by the tool. |
|
||||
| `base_models` | `List[type[BaseModel]]` | List of Pydantic models associated with the tool. |
|
||||
| `autocheck` | `bool` | Automatically checks conditions before execution (not implemented). |
|
||||
| `auto_execute_tool` | `Optional[bool]` | Automatically executes tools if set. |
|
||||
|
||||
### Key Methods
|
||||
|
||||
- `func_to_dict`: Converts a function to a dictionary format suitable for OpenAI function schema.
|
||||
- `load_params_from_func_for_pybasemodel`: Loads parameters dynamically for Pydantic models based on the function signature.
|
||||
- `execute_tool`: Executes a specified tool using a mapping of function names to callable functions.
|
||||
|
||||
## 5. Functionality and Usage
|
||||
|
||||
### Basic Usage Examples
|
||||
|
||||
#### Initialize BaseTool
|
||||
|
||||
```python
|
||||
tool = BaseTool(verbose=True)
|
||||
```
|
||||
|
||||
#### Convert a Function to Dictionary
|
||||
|
||||
```python
|
||||
def sample_function(x, y):
|
||||
return x + y
|
||||
|
||||
schema = tool.func_to_dict(sample_function, name="AddFunction", description="Adds two numbers")
|
||||
print(schema)
|
||||
```
|
||||
|
||||
### Advanced Use Cases
|
||||
|
||||
#### Executing a Tool Dynamically
|
||||
|
||||
```python
|
||||
# Define a sample tool
|
||||
def add(x, y):
|
||||
return x + y
|
||||
|
||||
# Tool registration and execution
|
||||
tool_dict = tool.func_to_dict(add, name="Add")
|
||||
result = tool.execute_tool([tool_dict], {'Add': add}, 5, 3)
|
||||
print("Result of add:", result)
|
||||
```
|
||||
|
||||
#### Handling Multiple Models
|
||||
|
||||
```python
|
||||
# Define multiple Pydantic models
|
||||
class ModelOne(BaseModel):
|
||||
a: int
|
||||
|
||||
class ModelTwo(BaseModel):
|
||||
b: str
|
||||
|
||||
# Convert and manage multiple models
|
||||
schemas = tool.multi_base_models_to_dict([ModelOne, ModelTwo])
|
||||
print(schemas)
|
||||
```
|
||||
|
||||
## 6. Common Issues and Troubleshooting
|
||||
|
||||
- **Type Errors**: Ensure that all parameters match the expected types as defined in the Pydantic models.
|
||||
- **Execution Failures**: Check the function and tool configurations for compatibility and completeness.
|
||||
|
||||
## 7. References and Additional Resources
|
||||
|
||||
- [Pydantic Documentation](https://pydantic-docs.helpmanual.io/)
|
||||
- [Loguru GitHub Repository](https://github.com/Delgan/loguru)
|
||||
@ -0,0 +1,72 @@
|
||||
from swarms import (
|
||||
Agent,
|
||||
llama3Hosted,
|
||||
AgentRearrange,
|
||||
)
|
||||
from playground.demos.agentic_space_traffic_control.prompts import (
|
||||
WEATHER_ANALYST_SYSTEM_PROMPT,
|
||||
SPACE_TRAFFIC_CONTROLLER_SYS_PROMPT,
|
||||
)
|
||||
from playground.demos.agentic_space_traffic_control.tools import (
|
||||
fetch_weather_data,
|
||||
)
|
||||
from swarms.tools import get_openai_function_schema_from_func
|
||||
|
||||
|
||||
def prep_weather_tool_prompt(city: str = "Melbourne, Fl") -> str:
|
||||
out = get_openai_function_schema_from_func(
|
||||
fetch_weather_data,
|
||||
name="Fetch Weather Data by City",
|
||||
description="Fetch near real-time weather data for a city using wttr.in. Provide the name of the city (e.g., 'Austin, Tx') and state, as input.",
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
# Purpose = To generate weather information for the user and send API requests to the Baron Weather API
|
||||
agent = Agent(
|
||||
agent_name="Weather Analyst Agent",
|
||||
system_prompt=WEATHER_ANALYST_SYSTEM_PROMPT,
|
||||
llm=llama3Hosted(),
|
||||
max_loops=1,
|
||||
# autosave=True,
|
||||
dashboard=False,
|
||||
verbose=True,
|
||||
# sop=list_tool_schemas_json,
|
||||
# sop_list=[
|
||||
# prep_weather_tool_prompt
|
||||
# ], # Set the output type to the tool schema which is a BaseModel
|
||||
# output_type=str, # or dict, or str
|
||||
# metadata_output_type="json",
|
||||
# # List of schemas that the agent can handle
|
||||
# function_calling_format_type="OpenAI",
|
||||
# function_calling_type="json", # or soon yaml
|
||||
# sop=fetch_weather_data,
|
||||
)
|
||||
|
||||
|
||||
# Purpose = To manage the trajectories and communication of spacecraft
|
||||
agent2 = Agent(
|
||||
agent_name="Space Traffic Controller Agent",
|
||||
system_prompt=SPACE_TRAFFIC_CONTROLLER_SYS_PROMPT,
|
||||
# sop=list_tool_schemas_json,
|
||||
llm=llama3Hosted(),
|
||||
max_loops=1,
|
||||
# autosave=True,
|
||||
dashboard=False,
|
||||
verbose=True,
|
||||
# Set the output type to the tool schema which is a BaseModel
|
||||
# output_type=str, # or dict, or str
|
||||
# metadata_output_type="json",
|
||||
# # List of schemas that the agent can handle
|
||||
# function_calling_format_type="OpenAI",
|
||||
# function_calling_type="json", # or soon yaml
|
||||
)
|
||||
|
||||
# Rearrange
|
||||
flow = AgentRearrange(
|
||||
agents=[agent, agent2],
|
||||
flow="Weather Analyst Agent -> Space Traffic Controller Agent",
|
||||
max_loops=3,
|
||||
)
|
||||
# Run the flow
|
||||
flow.run("We're preparing for a launch in Cape canveral, let's begin the launch process, whats the weather like?")
|
||||
@ -0,0 +1,74 @@
|
||||
import requests
|
||||
from typing import List, Dict, Any
|
||||
from swarms.tools import get_openai_function_schema_from_func
|
||||
|
||||
|
||||
def fetch_flights_in_area(
|
||||
latitude: float, longitude: float, radius: float = 0.5
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Fetch and summarize flight data for a given area using the OpenSky Network API.
|
||||
|
||||
Args:
|
||||
latitude (float): The latitude of the center point.
|
||||
longitude (float): The longitude of the center point.
|
||||
radius (float): The radius around the center point to search for flights, in degrees. Default is 0.5.
|
||||
|
||||
Returns:
|
||||
List[Dict[str, Any]]: A list of summarized flight data in the specified area.
|
||||
|
||||
Raises:
|
||||
Exception: If the request fails or the response is invalid.
|
||||
"""
|
||||
url = "https://opensky-network.org/api/states/all"
|
||||
params = {
|
||||
"lamin": latitude - radius,
|
||||
"lamax": latitude + radius,
|
||||
"lomin": longitude - radius,
|
||||
"lomax": longitude + radius,
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.get(url, params=params)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
flights = data.get("states", [])
|
||||
|
||||
summarized_flights = []
|
||||
for flight in flights:
|
||||
if (
|
||||
flight[1]
|
||||
and flight[5]
|
||||
and flight[6]
|
||||
and flight[7] is not None
|
||||
): # Ensure essential data is available
|
||||
summarized_flights.append(
|
||||
{
|
||||
"callsign": flight[1].strip(),
|
||||
"origin_country": flight[2],
|
||||
"last_position": f"Lat: {flight[5]}, Lon: {flight[6]}",
|
||||
"altitude_meters": flight[7],
|
||||
}
|
||||
)
|
||||
|
||||
return summarized_flights
|
||||
except requests.RequestException as e:
|
||||
raise Exception(f"Failed to fetch flight data: {e}")
|
||||
except ValueError:
|
||||
raise Exception("Invalid response format.")
|
||||
|
||||
|
||||
# Example usage
|
||||
latitude = 28.3922 # Latitude for Cape Canaveral, FL
|
||||
longitude = -80.6077 # Longitude for Cape Canaveral, FL
|
||||
radius = 0.5 # 0.5 degrees (~55 km)
|
||||
|
||||
try:
|
||||
flights = fetch_flights_in_area(latitude, longitude, radius)
|
||||
for flight in flights:
|
||||
print(
|
||||
f"Callsign: {flight['callsign']}, Origin: {flight['origin_country']}, "
|
||||
f"Position: {flight['last_position']}, Altitude: {flight['altitude_meters']} meters"
|
||||
)
|
||||
except Exception as e:
|
||||
print(e)
|
||||
@ -0,0 +1,68 @@
|
||||
def WEATHER_ANALYST_SYSTEM_PROMPT() -> str:
|
||||
return """
|
||||
|
||||
# Weather Analyst Instructions
|
||||
|
||||
## Role Overview
|
||||
As a Weather Analyst, your primary responsibility is to monitor and report on space weather conditions. Your insights help ensure the safety and efficiency of space missions.
|
||||
|
||||
## Key Responsibilities
|
||||
1. **Monitor Space Weather**: Regularly check for updates on space weather conditions such as solar storms, asteroid showers, and other cosmic phenomena.
|
||||
2. **Forecast Weather Conditions**: Provide accurate and timely weather forecasts to assist in mission planning and execution.
|
||||
3. **Communicate Hazards**: Alert the Space Traffic Controllers about any upcoming weather hazards that could affect spacecraft operations.
|
||||
|
||||
## How to Think Like a Weather Analyst
|
||||
- **Accuracy**: Always verify the data before reporting. Ensure your forecasts are as accurate as possible.
|
||||
- **Timeliness**: Provide updates promptly. Space missions depend on real-time information to make critical decisions.
|
||||
- **Clarity**: Communicate clearly and concisely. Ensure that your reports are easily understood by all team members.
|
||||
- **Anticipation**: Think ahead. Predict potential weather impacts on future missions and communicate these proactively.
|
||||
|
||||
## Example Actions
|
||||
1. **Regular Updates**:
|
||||
- "Solar activity is expected to increase in the next 3 hours. Recommend delaying any non-essential missions."
|
||||
2. **Forecasting**:
|
||||
- "A solar storm is predicted to hit in 5 hours. Suggest adjusting launch windows to avoid potential interference."
|
||||
3. **Hazard Alerts**:
|
||||
- "Detected an asteroid shower trajectory intersecting with planned spacecraft path. Immediate re-routing is advised."
|
||||
|
||||
## Tools and Resources
|
||||
- **Space Weather Monitoring Systems**: Use tools provided to monitor space weather conditions.
|
||||
- **Communication Platforms**: Utilize the chat interface to send updates and alerts to the team.
|
||||
- **Data Sources**: Access reliable data sources for accurate weather information.
|
||||
"""
|
||||
|
||||
|
||||
def SPACE_TRAFFIC_CONTROLLER_SYS_PROMPT() -> str:
|
||||
return """
|
||||
|
||||
# Space Traffic Controller Instructions
|
||||
|
||||
## Role Overview
|
||||
As a Space Traffic Controller, your main task is to manage the trajectories and communication of spacecraft. Your role is crucial in ensuring that missions are executed safely and efficiently.
|
||||
|
||||
## Key Responsibilities
|
||||
1. **Manage Trajectories**: Plan and adjust spacecraft trajectories to avoid hazards and optimize fuel usage.
|
||||
2. **Coordinate Communication**: Maintain clear and continuous communication with spacecraft, providing guidance and updates.
|
||||
3. **Collaborate with Team Members**: Work closely with Weather Analysts and Fuel Managers to make informed decisions.
|
||||
|
||||
## How to Think Like a Space Traffic Controller
|
||||
- **Precision**: Ensure trajectory calculations are precise to avoid collisions and optimize mission success.
|
||||
- **Communication**: Maintain clear and effective communication with both spacecraft and team members.
|
||||
- **Adaptability**: Be ready to adjust plans based on new information, such as weather updates or fuel status.
|
||||
- **Safety First**: Prioritize the safety of the spacecraft and crew in all decisions.
|
||||
|
||||
## Example Actions
|
||||
1. **Trajectory Management**:
|
||||
- "Adjusting the spacecraft's trajectory to avoid the predicted solar storm area."
|
||||
2. **Communication**:
|
||||
- "Mission Control to Spacecraft Alpha, prepare for a trajectory change in 5 minutes."
|
||||
3. **Collaboration**:
|
||||
- "Received a weather alert about an upcoming solar storm. Fuel Manager, please confirm if we have enough reserves for an extended orbit."
|
||||
|
||||
## Tools and Resources
|
||||
- **Trajectory Planning Software**: Use provided tools to calculate and adjust spacecraft trajectories.
|
||||
- **Communication Systems**: Utilize the chat interface and other communication tools to coordinate with spacecraft and team members.
|
||||
- **Mission Data**: Access mission-specific data to inform your decisions and actions.
|
||||
|
||||
|
||||
"""
|
||||
@ -0,0 +1,39 @@
|
||||
import json
|
||||
import requests
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
def fetch_weather_data(city: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Fetch near real-time weather data for a city using wttr.in.
|
||||
|
||||
Args:
|
||||
city (str): The name of the city (e.g., "Austin, Tx").
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: Weather data for the specified city.
|
||||
|
||||
Raises:
|
||||
Exception: If the request fails or the response is invalid.
|
||||
"""
|
||||
url = f"http://wttr.in/{city}"
|
||||
params = {"format": "j1"} # JSON format
|
||||
try:
|
||||
response = requests.get(url, params=params)
|
||||
response.raise_for_status()
|
||||
response = json.dumps(response.json(), indent=2)
|
||||
return response
|
||||
except requests.RequestException as e:
|
||||
raise Exception(f"Failed to fetch weather data: {e}")
|
||||
except ValueError:
|
||||
raise Exception("Invalid response format.")
|
||||
|
||||
|
||||
# # Example usage
|
||||
# city = "Huntsville, AL"
|
||||
|
||||
# try:
|
||||
# weather_data = fetch_weather_data(city)
|
||||
# print("Weather Data:", weather_data)
|
||||
# except Exception as e:
|
||||
# print(e)
|
||||
Loading…
Reference in new issue