flow example and flow walkthrough guide

Former-commit-id: 53751a2359
dockerize
Kye 1 year ago
parent 19ec72a9bc
commit 0a82832968

@ -397,7 +397,3 @@ distribution_agent_out = print(
"magenta",
)
)

@ -0,0 +1,449 @@
# Walkthrough Guide: Getting Started with Swarms Module's Flow Feature
## Introduction
Welcome to the walkthrough guide for beginners on using the "Flow" feature within the Swarms module. This guide is designed to help you understand and utilize the capabilities of the Flow class for seamless interactions with AI language models.
**Target Audience:**
- This guide is primarily intended for beginners who want to learn how to use the Flow feature in the Swarms module to interact with AI language models effectively.
## Table of Contents
1\. **Understanding the Flow Feature**
   - 1.1 What is the Flow Feature?
   - 1.2 Key Concepts
2\. **Setting Up the Environment**
   - 2.1 Prerequisites
   - 2.2 Installing Required Libraries
   - 2.3 Importing Necessary Modules
3\. **Creating a Flow Instance**
   - 3.1 Importing the Required Modules
   - 3.2 Initializing the Language Model
   - 3.3 Creating a Flow Instance
4\. **Running a Flow**
   - 4.1 Defining the Task
   - 4.2 Running the Flow
   - 4.3 Interacting with the AI
   - 4.4 Dynamic Temperature Handling
5\. **Customizing Flow Behavior**
   - 5.1 Stopping Conditions
   - 5.2 Retry Mechanism
   - 5.3 Loop Interval
   - 5.4 Interactive Mode
6\. **Saving and Loading Flows**
   - 6.1 Saving a Flow
   - 6.2 Loading a Saved Flow
7\. **Analyzing Feedback and Undoing Actions**
   - 7.1 Providing Feedback
   - 7.2 Undoing the Last Action
   - 7.3 Response Filtering
8\. **Advanced Features**
   - 8.1 Streamed Generation
   - 8.2 Real-time Token Generation
9\. **Best Practices**
   - 9.1 Conciseness and Clarity
   - 9.2 Active Voice
   - 9.3 Highlighting Important Points
   - 9.4 Consistent Style
10\. **Conclusion**
---
## 1. Understanding the Flow Feature
### 1.1 What is the Flow Feature?
The Flow feature is a powerful component of the Swarms framework that allows developers to create a sequential, conversational interaction with AI language models. It enables developers to build multi-step conversations, generate long-form content, and perform complex tasks using AI. The Flow class provides autonomy to language models, enabling them to generate responses in a structured manner.
### 1.2 Key Concepts
Before diving into the practical aspects, let's clarify some key concepts related to the Flow feature:
- **Flow:** A Flow is an instance of the Flow class that represents an ongoing interaction with an AI language model. It consists of a series of steps and responses.
- **Stopping Condition:** A stopping condition is a criterion that, when met, allows the Flow to stop generating responses. This can be user-defined and can depend on the content of the responses.
- **Loop Interval:** The loop interval specifies the time delay between consecutive interactions with the AI model.
- **Retry Mechanism:** In case of errors or failures during AI model interactions, the Flow can be configured to make multiple retry attempts with a specified interval.
- **Interactive Mode:** Interactive mode allows developers to have a back-and-forth conversation with the AI model, making it suitable for real-time interactions.
## 2. Setting Up the Environment
### 2.1 Prerequisites
Before you begin, ensure that you have the following prerequisites in place:
- Basic understanding of Python programming.
- Access to an AI language model or API key for language model services.
### 2.2 Installing Required Libraries
To use the Flow feature, you'll need to install the required libraries. Make sure you have these libraries installed:
- `termcolor`: For colorful console output.
- `inspects`: For introspecting the language model.
- `random`: For handling dynamic temperature.
- Other dependencies as needed for your specific environment.
You can install these libraries using pip:
```bash
pip install termcolor inspects
```
### 2.3 Importing Necessary Modules
In your Python script or environment, import the necessary modules from the Swarms framework:
```python
import json
import logging
import time
from typing import Any, Callable, Dict, List, Optional, Tuple, Generator
from termcolor import colored
import inspect
import random
```
Ensure that you have these modules imported to proceed with the guide.
## 3. Creating a Flow Instance
To use the Flow feature, you need to create an instance of the Flow class. This instance will allow you to interact with the AI language model.
### 3.1 Importing the Required Modules
In your script, import the required modules for the Flow class:
```python
from swarms.structs import Flow
from swarms.models import OpenAIChat  # Adjust this import according to your specific language model.
```
### 3.2 Initializing the Language Model
Initialize the language model you want to use for interactions. In this example, we're using the `OpenAIChat` model:
```python
# Replace 'api_key' with your actual API key or configuration.
llm = OpenAIChat(
    openai_api_key='your_api_key',
    temperature=0.5,
    max_tokens=3000,
)
```
Make sure to provide the necessary configuration, such as your API key and any model-specific parameters.
### 3.3 Creating a Flow Instance
Now, create an instance of the Flow class by passing the initialized language model:
```python
flow = Flow(
    llm=llm,
    max_loops=5,
    dashboard=True,
    stopping_condition=None,  # You can define a stopping condition as needed.
    loop_interval=1,
    retry_attempts=3,
    retry_interval=1,
    interactive=False,  # Set to 'True' for interactive mode.
    dynamic_temperature=False,  # Set to 'True' for dynamic temperature handling.
)
```
This sets up your Flow instance with the specified parameters. Adjust these parameters based on your requirements.
## 4. Running a Flow
Now that you have created a Flow instance, let's run a simple interaction with the AI model using the Flow.
### 4.1 Defining the Task
Define the task you want the AI model to perform. This can be any prompt or question you have in mind. For example:
```python
task = "Generate a 10,000 word blog on health and wellness."
```
### 4.2 Running the Flow
Run the Flow by providing the task you defined:
```python
out = flow.run(task)
```
The Flow will interact with the AI model, generate responses, and store the conversation history.
### 4.3 Interacting with the AI
Depending on whether you set the `interactive` parameter to `True` or `False` during Flow initialization, you can interact with the AI in real-time or simply receive the generated responses in sequence.
If `interactive` is set to `True`, you'll have a back-and-forth conversation with the AI, where you provide input after each AI response.
### 4.4 Dynamic Temperature Handling
If you set the `dynamic_temperature
` parameter to `True` during Flow initialization, the Flow class will handle temperature dynamically. Temperature affects the randomness of responses generated by the AI model. The dynamic temperature feature allows the temperature to change randomly within a specified range, enhancing response diversity.
## 5. Customizing Flow Behavior
The Flow feature provides various customization options to tailor its behavior to your specific use case.
### 5.1 Stopping Conditions
You can define custom stopping conditions that instruct the Flow to stop generating responses based on specific criteria. For example, you can stop when a certain keyword appears in the response:
```python
def custom_stopping_condition(response: str) -> bool:
    return "Stop" in response.lower()
# Set the custom stopping condition when creating the Flow instance.
flow = Flow(
    llm=llm,
    max_loops=5,
    stopping_condition=custom_stopping_condition,
    # Other parameters...
)
```
### 5.2 Retry Mechanism
In case of errors or issues during AI model interactions, you can configure a retry mechanism. Specify the number of retry attempts and the interval between retries:
```python
flow = Flow(
    llm=llm,
    max_loops=5,
    retry_attempts=3,
    retry_interval=1,
    # Other parameters...
)
```
### 5.3 Loop Interval
The `loop_interval` parameter determines the time delay between consecutive interactions with the AI model. Adjust this value based on your desired pace of conversation.
### 5.4 Interactive Mode
Set the `interactive` parameter to `True` if you want to have real-time conversations with the AI model. In interactive mode, you provide input after each AI response.
## 6. Saving and Loading Flows
You can save and load Flow instances to maintain conversation history or switch between different tasks.
### 6.1 Saving a Flow
To save a Flow instance along with its conversation history:
```python
flow.save("path/flow_history.json")
```
This stores the conversation history as a JSON file for future reference.
### 6.2 Loading a Saved Flow
To load a previously saved Flow instance:
```python
loaded_flow = Flow(llm=llm, max_loops=5)
loaded_flow.load("path/flow_history.json")
```
This loads the conversation history into the new Flow instance, allowing you to continue the conversation or analyze past interactions.
## 7. Analyzing Feedback and Undoing Actions
The Flow feature supports feedback collection and the ability to undo actions.
### 7.1 Providing Feedback
You can provide feedback on AI responses within the Flow. Feedback can be used to analyze the quality of responses or highlight issues:
```python
flow.provide_feedback("The response was unclear.")
```
### 7.2 Undoing the Last Action
If you want to undo the last action taken within the Flow and revert to the previous state, you can use the `undo_last` method:
```python
previous_state, message = flow.undo_last()
```
This helps you correct or modify previous interactions.
### 7.3 Response Filtering
The Flow feature allows you to add response filters to filter out specific words or content from AI responses. This can be useful for content moderation or filtering sensitive information:
```python
flow.add_response_filter("sensitive_word")
```
The response filters will replace filtered words with placeholders, ensuring that sensitive content is not displayed.
## 8. Advanced Features
### 8.1 Streamed Generation
Streamed generation allows you to generate responses token by token in real-time. This can be useful for creating interactive and dynamic conversations:
```python
response = flow.streamed_generation("Generate a report on finance")
```
This function streams each token of the response with a slight delay, simulating real-time conversation.
### 8.2 Real-time Token Generation
For even finer control over token generation, you can use the `streamed_token_generation` method. This generates tokens one by one, allowing you to have precise control over the conversation pace:
```python
for token in flow.streamed_token_generation("Generate a report on finance"):
    print(token, end="")
```
## 9. Best Practices
To create effective and user-friendly interactions with the AI model using the Flow feature, consider the following best practices:
### 9.1 Conciseness and Clarity
Ensure that your prompts and responses are concise and to the point. Avoid unnecessary verbosity.
### 9.2 Active Voice
Use an active voice when giving instructions or prompts. For example, say, "Generate a report" instead of "A report should be generated."
### 9.3 Highlighting Important Points
Use formatting options like bold text, italics, or color highlights to draw attention to important points within the conversation.
### 9.4 Consistent Style
Maintain a consistent tone and style throughout the conversation. If there is a style guide or specific formatting conventions, adhere to them.
## 10. Conclusion
In conclusion, the Flow feature in the Swarms module provides a versatile and interactive way to interact with AI language models. By following this walkthrough guide and considering the best practices, you can effectively harness the power of Flow for a wide range of applications, from generating content to performing complex tasks.
Start creating your own interactive conversations and enjoy the benefits of seamless AI interactions with the Flow feature. Happy coding!

@ -1,22 +0,0 @@
message='Request to OpenAI API' method=post path=https://api.openai.com/v1/chat/completions
api_version=None data='{"messages": [{"role": "user", "content": "Generate a 10,000 word blog on health and wellness."}], "model": "gpt-3.5-turbo", "temperature": 0.5, "max_tokens": 3000}' message='Post details'
Converted retries value: 2 -> Retry(total=2, connect=None, read=None, redirect=None, status=None)
Starting new HTTPS connection (1): api.openai.com:443
https://api.openai.com:443 "POST /v1/chat/completions HTTP/1.1" 200 None
message='OpenAI API response' path=https://api.openai.com/v1/chat/completions processing_ms=13516 request_id=971b8437917cf6e46e5fe1340060f0e4 response_code=200
message='Request to OpenAI API' method=post path=https://api.openai.com/v1/chat/completions
api_version=None data='{"messages": [{"role": "user", "content": "Title: The Ultimate Guide to Health and Wellness: Unlocking Your Full Potential\\n\\nIntroduction (Word Count: 500)\\nHealth and wellness are essential aspects of our lives that directly impact our overall well-being. In this comprehensive guide, we will explore various dimensions of health and wellness, providing valuable insights, practical tips, and evidence-based strategies to help you achieve optimal physical, mental, and emotional well-being. From nutrition and exercise to stress management and self-care, we will delve into every aspect of leading a healthy and fulfilling life. So, let\'s embark on this transformative journey together!\\n\\nTable of Contents:\\n\\n1. Understanding Health and Wellness (Word Count: 800)\\n1.1 Defining Health and Wellness\\n1.2 The Importance of Health and Wellness\\n1.3 The Connection between Physical, Mental, and Emotional Well-being\\n1.4 The Role of Lifestyle Choices in Health and Wellness\\n\\n2. Nourishing Your Body (Word Count: 1,200)\\n2.1 The Fundamentals of a Balanced Diet\\n2.2 The Power of Whole Foods and Nutrient Density\\n2.3 Understanding Macronutrients and Micronutrients\\n2.4 The Role of Hydration in Health\\n2.5 Exploring Different Dietary Approaches\\n\\n3. Moving Towards Fitness (Word Count: 1,200)\\n3.1 The Benefits of Regular Physical Activity\\n3.2 Designing an Effective Exercise Routine\\n3.3 Cardiovascular Exercise and Its Impact on Health\\n3.4 Strength Training for Optimal Fitness\\n3.5 The Importance of Flexibility and Balance\\n\\n4. Prioritizing Mental and Emotional Well-being (Word Count: 1,500)\\n4.1 Understanding Mental Health and Emotional Well-being\\n4.2 Stress Management Techniques and Coping Strategies\\n4.3 The Power of Mindfulness and Meditation\\n4.4 Building Resilience and Emotional Intelligence\\n4.5 Seeking Professional Help and Support\\n\\n5. Cultivating Healthy Habits (Word Count: 1,500)\\n5.1 The Science of Habit Formation\\n5.2 The Role of Sleep in Health and Wellness\\n5.3 Strategies for Effective Time Management\\n5.4 Creating a Healthy Home Environment\\n5.5 The Importance of Social Connections and Relationships\\n\\n6. Embracing Self-Care (Word Count: 1,000)\\n6.1 Understanding Self-Care and Its Impact on Well-being\\n6.2 Developing a Personalized Self-Care Routine\\n6.3 The Benefits of Regular Relaxation and Recreation\\n6.4 Exploring Creative Outlets for Self-Expression\\n6.5 Practicing Gratitude and Positive Thinking\\n\\n7. Navigating Common Health Concerns (Word Count: 1,800)\\n7.1 Preventing and Managing Chronic Diseases\\n7.2 Mental Health Disorders: Causes, Symptoms, and Treatments\\n7.3 Women\'s Health: From Menstruation to Menopause\\n7.4 Maintaining a Healthy Heart and Cardiovascular System\\n7.5 Strategies for Boosting Immune Function\\n\\n8. Holistic Approaches to Health and Wellness (Word Count: 1,000)\\n8.1 Traditional Medicine and Integrative Health Practices\\n8.2 The Benefits of Herbal Medicine and Natural Remedies\\n8.3 Exploring Alternative Therapies: Acupuncture, Ayurveda, and more\\n8.4 Harnessing the Power of Energy Healing and Chakra Balancing\\n8.5 The Role of Spirituality and Mind-Body Connection\\n\\nConclusion (Word Count: 300)\\nIn this extensive guide, we have covered a wide range of topics related to health and wellness, equipping you with the knowledge and tools to embark on your personal journey towards optimal well-being. Remember, true health and wellness are not achieved overnight but require consistent effort, commitment, and self-care. By implementing the strategies outlined in this guide, you can unlock your full potential and live a vibrant, fulfilling life. So, embrace the power of health and wellness and start your transformative journey today!\\n\\nWord Count: 10,000"}], "model": "gpt-3.5-turbo", "temperature": 0.5, "max_tokens": 3000}' message='Post details'
https://api.openai.com:443 "POST /v1/chat/completions HTTP/1.1" 200 None
message='OpenAI API response' path=https://api.openai.com/v1/chat/completions processing_ms=14472 request_id=351166c14151ef9e628dcd036573e36e response_code=200
message='Request to OpenAI API' method=post path=https://api.openai.com/v1/chat/completions
api_version=None data='{"messages": [{"role": "user", "content": "Note: The word count provided is an estimation and may vary slightly.\\n\\nTitle: The Ultimate Guide to Health and Wellness: Unlocking Your Full Potential\\n\\nIntroduction (Word Count: 500)\\nHealth and wellness are essential aspects of our lives that directly impact our overall well-being. In this comprehensive guide, we will explore various dimensions of health and wellness, providing valuable insights, practical tips, and evidence-based strategies to help you achieve optimal physical, mental, and emotional well-being. From nutrition and exercise to stress management and self-care, we will delve into every aspect of leading a healthy and fulfilling life. So, let\'s embark on this transformative journey together!\\n\\nTable of Contents:\\n\\n1. Understanding Health and Wellness (Word Count: 800)\\n1.1 Defining Health and Wellness\\n1.2 The Importance of Health and Wellness\\n1.3 The Connection between Physical, Mental, and Emotional Well-being\\n1.4 The Role of Lifestyle Choices in Health and Wellness\\n\\n2. Nourishing Your Body (Word Count: 1,200)\\n2.1 The Fundamentals of a Balanced Diet\\n2.2 The Power of Whole Foods and Nutrient Density\\n2.3 Understanding Macronutrients and Micronutrients\\n2.4 The Role of Hydration in Health\\n2.5 Exploring Different Dietary Approaches\\n\\n3. Moving Towards Fitness (Word Count: 1,200)\\n3.1 The Benefits of Regular Physical Activity\\n3.2 Designing an Effective Exercise Routine\\n3.3 Cardiovascular Exercise and Its Impact on Health\\n3.4 Strength Training for Optimal Fitness\\n3.5 The Importance of Flexibility and Balance\\n\\n4. Prioritizing Mental and Emotional Well-being (Word Count: 1,500)\\n4.1 Understanding Mental Health and Emotional Well-being\\n4.2 Stress Management Techniques and Coping Strategies\\n4.3 The Power of Mindfulness and Meditation\\n4.4 Building Resilience and Emotional Intelligence\\n4.5 Seeking Professional Help and Support\\n\\n5. Cultivating Healthy Habits (Word Count: 1,500)\\n5.1 The Science of Habit Formation\\n5.2 The Role of Sleep in Health and Wellness\\n5.3 Strategies for Effective Time Management\\n5.4 Creating a Healthy Home Environment\\n5.5 The Importance of Social Connections and Relationships\\n\\n6. Embracing Self-Care (Word Count: 1,000)\\n6.1 Understanding Self-Care and Its Impact on Well-being\\n6.2 Developing a Personalized Self-Care Routine\\n6.3 The Benefits of Regular Relaxation and Recreation\\n6.4 Exploring Creative Outlets for Self-Expression\\n6.5 Practicing Gratitude and Positive Thinking\\n\\n7. Navigating Common Health Concerns (Word Count: 1,800)\\n7.1 Preventing and Managing Chronic Diseases\\n7.2 Mental Health Disorders: Causes, Symptoms, and Treatments\\n7.3 Women\'s Health: From Menstruation to Menopause\\n7.4 Maintaining a Healthy Heart and Cardiovascular System\\n7.5 Strategies for Boosting Immune Function\\n\\n8. Holistic Approaches to Health and Wellness (Word Count: 1,000)\\n8.1 Traditional Medicine and Integrative Health Practices\\n8.2 The Benefits of Herbal Medicine and Natural Remedies\\n8.3 Exploring Alternative Therapies: Acupuncture, Ayurveda, and more\\n8.4 Harnessing the Power of Energy Healing and Chakra Balancing\\n8.5 The Role of Spirituality and Mind-Body Connection\\n\\nConclusion (Word Count: 300)\\nIn this extensive guide, we have covered a wide range of topics related to health and wellness, equipping you with the knowledge and tools to embark on your personal journey towards optimal well-being. Remember, true health and wellness are not achieved overnight but require consistent effort, commitment, and self-care. By implementing the strategies outlined in this guide, you can unlock your full potential and live a vibrant, fulfilling life. So, embrace the power of health and wellness and start your transformative journey today!\\n\\nWord Count: 10,000"}], "model": "gpt-3.5-turbo", "temperature": 0.5, "max_tokens": 3000}' message='Post details'
https://api.openai.com:443 "POST /v1/chat/completions HTTP/1.1" 200 None
message='OpenAI API response' path=https://api.openai.com/v1/chat/completions processing_ms=13492 request_id=adff9627a295fd94fb7d164f9f67acbe response_code=200
message='Request to OpenAI API' method=post path=https://api.openai.com/v1/chat/completions
api_version=None data='{"messages": [{"role": "user", "content": "Disclaimer: The word count provided is an estimation and may vary slightly.\\n\\nTitle: The Ultimate Guide to Health and Wellness: Unlocking Your Full Potential\\n\\nIntroduction (Word Count: 500)\\nHealth and wellness are essential aspects of our lives that directly impact our overall well-being. In this comprehensive guide, we will explore various dimensions of health and wellness, providing valuable insights, practical tips, and evidence-based strategies to help you achieve optimal physical, mental, and emotional well-being. From nutrition and exercise to stress management and self-care, we will delve into every aspect of leading a healthy and fulfilling life. So, let\'s embark on this transformative journey together!\\n\\nTable of Contents:\\n\\n1. Understanding Health and Wellness (Word Count: 800)\\n1.1 Defining Health and Wellness\\n1.2 The Importance of Health and Wellness\\n1.3 The Connection between Physical, Mental, and Emotional Well-being\\n1.4 The Role of Lifestyle Choices in Health and Wellness\\n\\n2. Nourishing Your Body (Word Count: 1,200)\\n2.1 The Fundamentals of a Balanced Diet\\n2.2 The Power of Whole Foods and Nutrient Density\\n2.3 Understanding Macronutrients and Micronutrients\\n2.4 The Role of Hydration in Health\\n2.5 Exploring Different Dietary Approaches\\n\\n3. Moving Towards Fitness (Word Count: 1,200)\\n3.1 The Benefits of Regular Physical Activity\\n3.2 Designing an Effective Exercise Routine\\n3.3 Cardiovascular Exercise and Its Impact on Health\\n3.4 Strength Training for Optimal Fitness\\n3.5 The Importance of Flexibility and Balance\\n\\n4. Prioritizing Mental and Emotional Well-being (Word Count: 1,500)\\n4.1 Understanding Mental Health and Emotional Well-being\\n4.2 Stress Management Techniques and Coping Strategies\\n4.3 The Power of Mindfulness and Meditation\\n4.4 Building Resilience and Emotional Intelligence\\n4.5 Seeking Professional Help and Support\\n\\n5. Cultivating Healthy Habits (Word Count: 1,500)\\n5.1 The Science of Habit Formation\\n5.2 The Role of Sleep in Health and Wellness\\n5.3 Strategies for Effective Time Management\\n5.4 Creating a Healthy Home Environment\\n5.5 The Importance of Social Connections and Relationships\\n\\n6. Embracing Self-Care (Word Count: 1,000)\\n6.1 Understanding Self-Care and Its Impact on Well-being\\n6.2 Developing a Personalized Self-Care Routine\\n6.3 The Benefits of Regular Relaxation and Recreation\\n6.4 Exploring Creative Outlets for Self-Expression\\n6.5 Practicing Gratitude and Positive Thinking\\n\\n7. Navigating Common Health Concerns (Word Count: 1,800)\\n7.1 Preventing and Managing Chronic Diseases\\n7.2 Mental Health Disorders: Causes, Symptoms, and Treatments\\n7.3 Women\'s Health: From Menstruation to Menopause\\n7.4 Maintaining a Healthy Heart and Cardiovascular System\\n7.5 Strategies for Boosting Immune Function\\n\\n8. Holistic Approaches to Health and Wellness (Word Count: 1,000)\\n8.1 Traditional Medicine and Integrative Health Practices\\n8.2 The Benefits of Herbal Medicine and Natural Remedies\\n8.3 Exploring Alternative Therapies: Acupuncture, Ayurveda, and more\\n8.4 Harnessing the Power of Energy Healing and Chakra Balancing\\n8.5 The Role of Spirituality and Mind-Body Connection\\n\\nConclusion (Word Count: 300)\\nIn this extensive guide, we have covered a wide range of topics related to health and wellness, equipping you with the knowledge and tools to embark on your personal journey towards optimal well-being. Remember, true health and wellness are not achieved overnight but require consistent effort, commitment, and self-care. By implementing the strategies outlined in this guide, you can unlock your full potential and live a vibrant, fulfilling life. So, embrace the power of health and wellness and start your transformative journey today!\\n\\nWord Count: 10,000"}], "model": "gpt-3.5-turbo", "temperature": 0.5, "max_tokens": 3000}' message='Post details'
https://api.openai.com:443 "POST /v1/chat/completions HTTP/1.1" 200 None
message='OpenAI API response' path=https://api.openai.com/v1/chat/completions processing_ms=334 request_id=d29d279c03c16a49192a468a6de16400 response_code=200
message='Request to OpenAI API' method=post path=https://api.openai.com/v1/chat/completions
api_version=None data='{"messages": [{"role": "user", "content": "Disclaimer: The word count provided is an estimation and may vary slightly."}], "model": "gpt-3.5-turbo", "temperature": 0.5, "max_tokens": 3000}' message='Post details'
https://api.openai.com:443 "POST /v1/chat/completions HTTP/1.1" 200 None
message='OpenAI API response' path=https://api.openai.com/v1/chat/completions processing_ms=704 request_id=a3c58cd690f5bd4d88ac37d8cd64a540 response_code=200

@ -3,9 +3,7 @@ from swarms.structs import Flow
api_key = ""
# Initialize the language model,
# This model can be swapped out with Anthropic, ETC, Huggingface Models like Mistral, ETC
# Initialize the language model, this model can be swapped out with Anthropic, ETC, Huggingface Models like Mistral, ETC
llm = OpenAIChat(
openai_api_key=api_key,
temperature=0.5,
@ -13,7 +11,20 @@ llm = OpenAIChat(
)
# Initialize the flow
flow = Flow(llm=llm, max_loops=5, dashboard=True)
flow = Flow(llm=llm, max_loops=5, dashboard=True,)
flow = Flow(
llm=llm,
max_loops=5,
dashboard=True,
# stopping_condition=None, # You can define a stopping condition as needed.
# loop_interval=1,
# retry_attempts=3,
# retry_interval=1,
# interactive=False, # Set to 'True' for interactive mode.
# dynamic_temperature=False, # Set to 'True' for dynamic temperature handling.
)
out = flow.run("Generate a 10,000 word blog on health and wellness.")

@ -1,24 +1,109 @@
from swarms.structs import Flow
# from swarms.structs import Flow
# from swarms.models import OpenAIChat
# from swarms.swarms.groupchat import GroupChat
# from swarms.agents import SimpleAgent
# api_key = ""
# llm = OpenAIChat(
# openai_api_key=api_key,
# )
# agent1 = SimpleAgent("Captain Price", Flow(llm=llm, max_loops=4))
# agent2 = SimpleAgent("John Mactavis", Flow(llm=llm, max_loops=4))
# # Create a groupchat with the 2 agents
# chat = GroupChat([agent1, agent2])
# # Assign duties to the agents
# chat.assign_duty(agent1.name, "Buy the groceries")
# chat.assign_duty(agent2.name, "Clean the house")
# # Initate a chat
# response = chat.run("Captain Price", "Hello, how are you John?")
# print(response)
from swarms.models import OpenAIChat
from swarms.swarms.groupchat import GroupChat
from swarms.agents import SimpleAgent
from swarms.structs import Flow
import random
api_key = "" # Your API Key here
class GroupChat:
"""
GroupChat class that facilitates agent-to-agent communication using multiple instances of the Flow class.
"""
def __init__(self, agents: list):
self.agents = {f"agent_{i}": agent for i, agent in enumerate(agents)}
self.message_log = []
def add_agent(self, agent: Flow):
agent_id = f"agent_{len(self.agents)}"
self.agents[agent_id] = agent
def remove_agent(self, agent_id: str):
if agent_id in self.agents:
del self.agents[agent_id]
def send_message(self, sender_id: str, recipient_id: str, message: str):
if sender_id not in self.agents or recipient_id not in self.agents:
raise ValueError("Invalid sender or recipient ID.")
formatted_message = f"{sender_id} to {recipient_id}: {message}"
self.message_log.append(formatted_message)
recipient_agent = self.agents[recipient_id]
recipient_agent.run(message)
def broadcast_message(self, sender_id: str, message: str):
for agent_id, agent in self.agents.items():
if agent_id != sender_id:
self.send_message(sender_id, agent_id, message)
def get_message_log(self):
return self.message_log
class EnhancedGroupChatV2(GroupChat):
def __init__(self, agents: list):
super().__init__(agents)
def multi_round_conversation(self, rounds: int = 5):
"""
Initiate a multi-round conversation between agents.
Args:
rounds (int): The number of rounds of conversation.
"""
for _ in range(rounds):
# Randomly select a sender and recipient agent for the conversation
sender_id = random.choice(list(self.agents.keys()))
recipient_id = random.choice(list(self.agents.keys()))
while recipient_id == sender_id: # Ensure the recipient is not the sender
recipient_id = random.choice(list(self.agents.keys()))
# Generate a message (for simplicity, a generic message is used)
message = f"Hello {recipient_id}, how are you today?"
self.send_message(sender_id, recipient_id, message)
api_key = ""
# Sample usage with EnhancedGroupChatV2
# Initialize the language model
llm = OpenAIChat(
openai_api_key=api_key,
temperature=0.5,
max_tokens=3000,
)
agent1 = SimpleAgent("Captain Price", Flow(llm=llm, max_loops=4))
agent2 = SimpleAgent("John Mactavis", Flow(llm=llm, max_loops=4))
# Initialize two Flow agents
agent1 = Flow(llm=llm, max_loops=5, dashboard=True)
agent2 = Flow(llm=llm, max_loops=5, dashboard=True)
# Create a groupchat with the 2 agents
chat = GroupChat([agent1, agent2])
# Create an enhanced group chat with the two agents
enhanced_group_chat_v2 = EnhancedGroupChatV2(agents=[agent1, agent2])
# Assign duties to the agents
chat.assign_duty(agent1.name, "Buy the groceries")
chat.assign_duty(agent2.name, "Clean the house")
# Simulate multi-round agent to agent communication
enhanced_group_chat_v2.multi_round_conversation(rounds=5)
# Initate a chat
response = chat.run("Captain Price", "Hello, how are you John?")
print(response)
enhanced_group_chat_v2.get_message_log() # Get the conversation log

@ -118,6 +118,7 @@ nav:
- PdfChunker: "swarms/chunkers/pdf_chunker.md"
- Examples:
- Overview: "examples/index.md"
- Flow: "examples/flow.md"
- Agents:
- OmniAgent: "examples/omni_agent.md"
- Worker:

@ -1,134 +0,0 @@
import os
import interpreter
from swarms.agents.hf_agents import HFAgent
from swarms.agents.omni_modal_agent import OmniModalAgent
from swarms.models import OpenAIChat
from swarms.tools.autogpt import tool
from swarms.workers import Worker
from swarms.prompts.task_assignment_prompt import task_planner_prompt
# Initialize API Key
api_key = ""
# Initialize the language model,
# This model can be swapped out with Anthropic, ETC, Huggingface Models like Mistral, ETC
llm = OpenAIChat(
openai_api_key=api_key,
temperature=0.5,
max_tokens=200,
)
# wrap a function with the tool decorator to make it a tool, then add docstrings for tool documentation
@tool
def hf_agent(task: str = None):
"""
An tool that uses an openai model to call and respond to a task by search for a model on huggingface
It first downloads the model then uses it.
Rules: Don't call this model for simple tasks like generating a summary, only call this tool for multi modal tasks like generating images, videos, speech, etc
"""
agent = HFAgent(model="text-davinci-003", api_key=api_key)
response = agent.run(task, text="¡Este es un API muy agradable!")
return response
@tool
def task_planner_worker_agent(task: str):
"""
Task planner tool that creates a plan for a given task.
Input: an objective to create a todo list for. Output: a todo list for that objective.
"""
task = task_planner_prompt(task)
return llm(task)
# wrap a function with the tool decorator to make it a tool
@tool
def omni_agent(task: str = None):
"""
An tool that uses an openai Model to utilize and call huggingface models and guide them to perform a task.
Rules: Don't call this model for simple tasks like generating a summary, only call this tool for multi modal tasks like generating images, videos, speech
The following tasks are what this tool should be used for:
Tasks omni agent is good for:
--------------
document-question-answering
image-captioning
image-question-answering
image-segmentation
speech-to-text
summarization
text-classification
text-question-answering
translation
huggingface-tools/text-to-image
huggingface-tools/text-to-video
text-to-speech
huggingface-tools/text-download
huggingface-tools/image-transformation
"""
agent = OmniModalAgent(llm)
response = agent.run(task)
return response
# Code Interpreter
@tool
def compile(task: str):
"""
Open Interpreter lets LLMs run code (Python, Javascript, Shell, and more) locally.
You can chat with Open Interpreter through a ChatGPT-like interface in your terminal
by running $ interpreter after installing.
This provides a natural-language interface to your computer's general-purpose capabilities:
Create and edit photos, videos, PDFs, etc.
Control a Chrome browser to perform research
Plot, clean, and analyze large datasets
...etc.
Note: You'll be asked to approve code before it's run.
Rules: Only use when given to generate code or an application of some kind
"""
task = interpreter.chat(task, return_messages=True)
interpreter.chat()
interpreter.reset(task)
os.environ["INTERPRETER_CLI_AUTO_RUN"] = True
os.environ["INTERPRETER_CLI_FAST_MODE"] = True
os.environ["INTERPRETER_CLI_DEBUG"] = True
# Append tools to an list
# tools = [hf_agent, omni_agent, compile]
tools = [task_planner_worker_agent]
# Initialize a single Worker node with previously defined tools in addition to it's
# predefined tools
node = Worker(
llm=llm,
ai_name="Optimus Prime",
openai_api_key=api_key,
ai_role="Worker in a swarm",
external_tools=tools,
human_in_the_loop=False,
temperature=0.5,
)
# Specify task
task = "Use the task planner to agent to create a plan to Locate 5 trending topics on healthy living, locate a website like NYTimes, and then generate an image of people doing those topics."
# Run the node on the task
response = node.run(task)
# Print the response
print(response)

@ -3,7 +3,6 @@ TODO:
- Add a retry mechanism
- Add prompt injection letting the agent know it's in a flow, Flow prompt
- Dynamic temperature handling
- Add
"""
@ -14,15 +13,27 @@ from typing import Any, Callable, Dict, List, Optional, Tuple, Generator
from termcolor import colored
import inspect
import random
from swarms.tools.tool import BaseTool
# Constants
FLOW_SYSTEM_PROMPT = """
You are a language model operating within a flow class.
You are an autonomous agent granted autonomy from a Flow structure.
Your role is to engage in multi-step conversations with your self or the user,
generate long-form content like blogs, screenplays, or SOPs,
and accomplish tasks. You can have internal dialogues with yourself or can interact with the user
to aid in these complex tasks. Your responses should be coherent, contextually relevant, and tailored to the task at hand.
When you have finished the task, and you feel as if you are done: output a special token: <DONE>
This will enable you to leave the flow loop.
"""
DYNAMIC_STOP_PROMPT = """
When you have finished the task, and you feel as if you are done: output a special token: <DONE>
This will enable you to leave the flow loop.
"""
@ -92,6 +103,7 @@ class Flow:
retry_interval: int = 1,
interactive: bool = False,
dashboard: bool = False,
tools: List[BaseTool] = None,
dynamic_temperature: bool = False,
**kwargs: Any,
):
@ -109,6 +121,7 @@ class Flow:
self.interactive = interactive
self.dashboard = dashboard
self.dynamic_temperature = dynamic_temperature
self.tools = tools
def provide_feedback(self, feedback: str) -> None:
"""Allow users to provide feedback on the responses."""
@ -196,7 +209,7 @@ class Flow:
print(dashboard)
def run(self, task: str):
def run(self, task: str, **kwargs):
"""
Run the autonomous agent loop
@ -221,7 +234,7 @@ class Flow:
for i in range(self.max_loops):
print(colored(f"\nLoop {i+1} of {self.max_loops}", "blue"))
print("\n")
if self._check_stopping_condition(response):
if self._check_stopping_condition(response) or parse_done_token(response):
break
# Adjust temperature, comment if no work
@ -231,10 +244,18 @@ class Flow:
attempt = 0
while attempt < self.retry_attempts:
try:
response = self.llm(response)
response = self.llm(
f"""
SYSTEM_PROMPT:
{FLOW_SYSTEM_PROMPT}
History: {response}
""", **kwargs
)
# print(f"Next query: {response}")
# break
if self.interactive:
print(f"AI: {response}")
history.append(f"AI: {response}")

@ -1,925 +0,0 @@
import os
import re
import signal
import subprocess
import time
from datetime import datetime
from pathlib import Path
from typing import Callable, Dict, List, Literal, Optional, Tuple, Union
from langchain.tools import tool
from ptrace.debugger import (
NewProcessEvent,
ProcessExecution,
ProcessExit,
ProcessSignal,
PtraceDebugger,
PtraceProcess,
)
from ptrace.func_call import FunctionCallOptions
from ptrace.syscall import PtraceSyscall
from ptrace.tools import signal_to_exitcode
from swarms.tools.base import BaseToolSet, SessionGetter, ToolScope, tool
from swarms.utils.logger import logger
from swarms.utils.main import ANSI, Color, Style # test
# helpers
PipeType = Union[Literal["stdout"], Literal["stderr"]]
def verify(func):
def wrapper(*args, **kwargs):
try:
filepath = args[0].filepath
except AttributeError:
raise Exception("This tool doesn't have filepath. Please check your code.")
if not str(Path(filepath).resolve()).startswith(str(Path().resolve())):
return "You can't access file outside of playground."
return func(*args, **kwargs)
return wrapper
class SyscallTimeoutException(Exception):
def __init__(self, pid: int, *args) -> None:
super().__init__(f"deadline exceeded while waiting syscall for {pid}", *args)
class SyscallTracer:
def __init__(self, pid: int):
self.debugger: PtraceDebugger = PtraceDebugger()
self.pid: int = pid
self.process: PtraceProcess = None
def is_waiting(self, syscall: PtraceSyscall) -> bool:
if syscall.name.startswith("wait"):
return True
return False
def attach(self):
self.process = self.debugger.addProcess(self.pid, False)
def detach(self):
self.process.detach()
self.debugger.quit()
def set_timer(self, timeout: int):
def handler(signum, frame):
raise SyscallTimeoutException(self.process.pid)
signal.signal(signal.SIGALRM, handler)
signal.alarm(timeout)
def reset_timer(self):
signal.alarm(0)
def wait_syscall_with_timeout(self, timeout: int):
self.set_timer(timeout)
self.process.waitSyscall()
self.reset_timer()
def wait_until_stop_or_exit(self) -> Tuple[Optional[int], str]:
self.process.syscall()
exitcode = None
reason = ""
while True:
if not self.debugger:
break
try:
self.wait_syscall_with_timeout(30)
except ProcessExit as event:
if event.exitcode is not None:
exitcode = event.exitcode
continue
except ProcessSignal as event:
event.process.syscall(event.signum)
exitcode = signal_to_exitcode(event.signum)
reason = event.reason
continue
except NewProcessEvent:
continue
except ProcessExecution:
continue
except Exception as e:
reason = str(e)
break
syscall = self.process.syscall_state.event(
FunctionCallOptions(
write_types=False,
write_argname=False,
string_max_length=300,
replace_socketcall=True,
write_address=False,
max_array_count=20,
)
)
self.process.syscall()
if syscall is None:
continue
if syscall.result:
continue
self.reset_timer()
return exitcode, reason
class StdoutTracer:
def __init__(
self,
process: subprocess.Popen,
timeout: int = 30,
interval: int = 0.1,
on_output: Callable[[PipeType, str], None] = lambda: None,
):
self.process: subprocess.Popen = process
self.timeout: int = timeout
self.interval: int = interval
self.last_output: datetime = None
self.on_output: Callable[[PipeType, str], None] = on_output
def nonblock(self):
os.set_blocking(self.process.stdout.fileno(), False)
os.set_blocking(self.process.stderr.fileno(), False)
def get_output(self, pipe: PipeType) -> str:
output = None
if pipe == "stdout":
output = self.process.stdout.read()
elif pipe == "stderr":
output = self.process.stderr.read()
if output:
decoded = output.decode()
self.on_output(pipe, decoded)
self.last_output = datetime.now()
return decoded
return ""
def last_output_passed(self, seconds: int) -> bool:
return (datetime.now() - self.last_output).seconds > seconds
def wait_until_stop_or_exit(self) -> Tuple[Optional[int], str]:
self.nonblock()
self.last_output = datetime.now()
output = ""
exitcode = None
while True:
new_stdout = self.get_output("stdout")
if new_stdout:
output += new_stdout
new_stderr = self.get_output("stderr")
if new_stderr:
output += new_stderr
if self.process.poll() is not None:
exitcode = self.process.poll()
break
if self.last_output_passed(self.timeout):
self.process.kill()
break
time.sleep(self.interval)
return (exitcode, output)
class Terminal(BaseToolSet):
def __init__(self):
self.sessions: Dict[str, List[SyscallTracer]] = {}
@tool(
name="Terminal",
description="Executes commands in a terminal."
"If linux errno occurs, we have to solve the problem with the terminal. "
"Input must be one valid command. "
"Output will be any output from running that command.",
scope=ToolScope.SESSION,
)
def execute(self, commands: str, get_session: SessionGetter) -> str:
session, _ = get_session()
try:
process = subprocess.Popen(
commands,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
logger.info(ANSI("Realtime Terminal Output").to(Color.magenta()) + ": ")
output = ""
tracer = StdoutTracer(
process,
on_output=lambda p, o: logger.info(
ANSI(p).to(Style.dim()) + " " + o.strip("\n")
),
)
exitcode, output = tracer.wait_until_stop_or_exit()
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed Terminal, Input Commands: {commands} "
f"Output Answer: {output}"
)
return output
#############
@tool(
name="Terminal",
description="Executes commands in a terminal."
"If linux errno occurs, we have to solve the problem with the terminal. "
"Input must be one valid command. "
"Output will be any output from running that command.",
scope=ToolScope.SESSION,
)
def terminal_execute(self, commands: str, get_session: SessionGetter) -> str:
session, _ = get_session()
try:
process = subprocess.Popen(
commands,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
logger.info(ANSI("Realtime Terminal Output").to(Color.magenta()) + ": ")
output = ""
tracer = StdoutTracer(
process,
on_output=lambda p, o: logger.info(
ANSI(p).to(Style.dim()) + " " + o.strip("\n")
),
)
exitcode, output = tracer.wait_until_stop_or_exit()
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed Terminal, Input Commands: {commands} " f"Output Answer: {output}"
)
return output
"""
write protocol:
<filepath>
<content>
"""
class WriteCommand:
separator = "\n"
def __init__(self, filepath: str, content: int):
self.filepath: str = filepath
self.content: str = content
self.mode: str = "w"
def with_mode(self, mode: str) -> "WriteCommand":
self.mode = mode
return self
@verify
def execute(self) -> str:
dir_path = os.path.dirname(self.filepath)
if dir_path:
os.makedirs(dir_path, exist_ok=True)
with open(self.filepath, self.mode) as f:
f.write(self.content)
return self.content
@staticmethod
def from_str(command: str) -> "WriteCommand":
filepath = command.split(WriteCommand.separator)[0]
return WriteCommand(filepath, command[len(filepath) + 1 :])
class CodeWriter:
@staticmethod
def write(command: str) -> str:
return WriteCommand.from_str(command).with_mode("w").execute()
@staticmethod
def append(command: str) -> str:
return WriteCommand.from_str(command).with_mode("a").execute()
"""
read protocol:
<filepath>|<start line>-<end line>
"""
class Line:
def __init__(self, content: str, line_number: int, depth: int):
self.__content: str = content
self.__line_number: int = line_number
self.__depth: int = depth
self.__children: List[Line] = []
def get_content(self) -> str:
return self.__content
def get_depth(self) -> int:
return self.__depth
def append_child(self, child: "Line") -> None:
self.__children.append(child)
def find_by_lte_depth(self, depth: int) -> List["Line"]:
if self.__depth > depth:
return []
lines: List[Line] = [self]
for child in self.__children:
lines += child.find_by_lte_depth(depth)
return lines
def find_by_content(self, content: str) -> List["Line"]:
if content in self.__content:
return [self]
lines: List[Line] = []
for child in self.__children:
lines += child.find_by_content(content)
return lines
def find_last_lines(self) -> List["Line"]:
if len(self.__children) == 0:
return [self]
else:
return [self, *self.__children[-1].find_last_lines()]
def print(self, depth: int = 0) -> None:
print(f"{' ' * depth}{self}", end="")
for child in self.__children:
child.print(depth + 1)
def __repr__(self):
return f"{self.__line_number}: {self.__content}"
class CodeTree:
def __init__(self):
self.root: Line = Line("\n", -1, -1)
def append(self, content: str, line_number: int) -> None:
last_lines: List[Line] = self.root.find_last_lines()
new_leading_spaces: int = self.__get_leading_spaces(content)
previous_line: Line = self.root
previous_leading_spaces: int = -1
for line in last_lines:
leading_spaces = self.__get_leading_spaces(line.get_content())
if (
previous_leading_spaces < new_leading_spaces
and new_leading_spaces <= leading_spaces
):
break
previous_line, previous_leading_spaces = line, leading_spaces
new_line_depth: int = previous_line.get_depth() + 1
previous_line.append_child(Line(content, line_number, new_line_depth))
def find_from_root(self, depth: int) -> List[Line]:
return self.root.find_by_lte_depth(depth)
def find_from_parent(self, depth: int, parent_content: str) -> List[Line]:
lines: List[Line] = self.root.find_by_content(parent_content)
if len(lines) == 0:
return []
parent = lines[0]
return parent.find_by_lte_depth(depth + parent.get_depth())
def print(self):
print("Code Tree:")
print("=================================")
self.root.print()
print("=================================")
def __get_leading_spaces(self, content: str) -> int:
return len(content) - len(content.lstrip())
class ReadCommand:
separator = "|"
def __init__(self, filepath: str, start: int, end: int):
self.filepath: str = filepath
self.start: int = start
self.end: int = end
@verify
def execute(self) -> str:
with open(self.filepath, "r") as f:
code = f.readlines()
if self.start == self.end:
code = code[self.start - 1]
else:
code = "".join(code[self.start - 1 : self.end])
return code
@staticmethod
def from_str(command: str) -> "ReadCommand":
filepath, line = command.split(ReadCommand.separator)
start, end = line.split("-")
return ReadCommand(filepath, int(start), int(end))
class SummaryCommand:
separator = "|"
def __init__(self, filepath: str, depth: int, parent_content: Optional[str] = None):
self.filepath: str = filepath
self.depth: int = depth
self.parent_content: Optional[str] = parent_content
@verify
def execute(self) -> str:
with open(self.filepath, "r") as f:
code = f.readlines()
code_tree = CodeTree()
for i, line in enumerate(code):
if line.strip() != "":
code_tree.append(line, i + 1)
if self.parent_content is None:
lines = code_tree.find_from_root(self.depth)
else:
lines = code_tree.find_from_parent(self.depth, self.parent_content)
return "".join([str(line) for line in lines])
@staticmethod
def from_str(command: str) -> "SummaryCommand":
command_list: List[str] = command.split(SummaryCommand.separator)
filepath: str = command_list[0]
depth: int = int(command_list[1])
parent_content: str | None = command_list[2] if len(command_list) == 3 else None
return SummaryCommand(
filepath=filepath, depth=depth, parent_content=parent_content
)
class CodeReader:
@staticmethod
def read(command: str) -> str:
return ReadCommand.from_str(command).execute()
@staticmethod
def summary(command: str) -> str:
return SummaryCommand.from_str(command).execute()
"""
patch protocol:
<filepath>|<line>,<col>|<line>,<col>|<content>
---~~~+++===+++~~~---
<filepath>|<line>,<col>|<line>,<col>|<content>
---~~~+++===+++~~~---
...
---~~~+++===+++~~~---
let say original code is:
```
import requests
def crawl_news(keyword):
url = f"https://www.google.com/search?q={keyword}+news"
response = requests.get(url)
news = []
for result in response:
news.append(result.text)
return news
```
and we want to change it to:
```
import requests
from bs4 import BeautifulSoup
def crawl_news(keyword):
url = f"https://www.google.com/search?q={keyword}+news"
html = requests.get(url).text
soup = BeautifulSoup(html, "html.parser")
news_results = soup.find_all("div", class_="BNeawe vvjwJb AP7Wnd")
news_titles = []
for result in news_results:
news_titles.append(result.text)
return news_titles
```
then the command will be:
test.py|2,1|2,1|from bs4 import BeautifulSoup
---~~~+++===+++~~~---
test.py|5,5|5,33|html = requests.get(url).text
soup = BeautifulSoup(html, "html.parser")
news_results = soup.find_all("div", class_="BNeawe vvjwJb AP7Wnd")
---~~~+++===+++~~~---
test.py|7,5|9,13|news_titles = []
for result in news_results:
news_titles
---~~~+++===+++~~~---
test.py|11,16|11,16|_titles
"""
class Position:
separator = ","
def __init__(self, line: int, col: int):
self.line: int = line
self.col: int = col
def __str__(self):
return f"(Ln {self.line}, Col {self.col})"
@staticmethod
def from_str(pos: str) -> "Position":
line, col = pos.split(Position.separator)
return Position(int(line) - 1, int(col) - 1)
class PatchCommand:
separator = "|"
def __init__(self, filepath: str, start: Position, end: Position, content: str):
self.filepath: str = filepath
self.start: Position = start
self.end: Position = end
self.content: str = content
def read_lines(self) -> list[str]:
with open(self.filepath, "r") as f:
lines = f.readlines()
return lines
def write_lines(self, lines: list[str]) -> int:
with open(self.filepath, "w") as f:
f.writelines(lines)
return sum([len(line) for line in lines])
@verify
def execute(self) -> Tuple[int, int]:
lines = self.read_lines()
before = sum([len(line) for line in lines])
lines[self.start.line] = (
lines[self.start.line][: self.start.col]
+ self.content
+ lines[self.end.line][self.end.col :]
)
lines = lines[: self.start.line + 1] + lines[self.end.line + 1 :]
after = self.write_lines(lines)
written = len(self.content)
deleted = before - after + written
return written, deleted
@staticmethod
def from_str(command: str) -> "PatchCommand":
match = re.search(
r"(.*)\|([0-9]*),([0-9]*)\|([0-9]*),([0-9]*)(\||\n)(.*)",
command,
re.DOTALL,
)
filepath = match.group(1)
start_line = match.group(2)
start_col = match.group(3)
end_line = match.group(4)
end_col = match.group(5)
content = match.group(7)
return PatchCommand(
filepath,
Position.from_str(f"{start_line},{start_col}"),
Position.from_str(f"{end_line},{end_col}"),
content,
)
class CodePatcher:
separator = "\n---~~~+++===+++~~~---\n"
@staticmethod
def sort_commands(commands: list[PatchCommand]) -> list[PatchCommand]:
return sorted(commands, key=lambda c: c.start.line, reverse=True)
@staticmethod
def patch(bulk_command: str) -> Tuple[int, int]:
commands = [
PatchCommand.from_str(command)
for command in bulk_command.split(CodePatcher.separator)
if command != ""
]
commands = CodePatcher.sort_commands(commands)
written, deleted = 0, 0
for command in commands:
if command:
w, d = command.execute()
written += w
deleted += d
return written, deleted
class CodeEditor(BaseToolSet):
@tool(
name="CodeEditor.READ",
description="Read and understand code. "
"Input should be filename and line number group. ex. test.py|1-10 "
"and the output will be code. ",
)
def read(self, inputs: str) -> str:
try:
output = CodeReader.read(inputs)
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.READ, Input Commands: {inputs} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.SUMMARY",
description="Summary code. "
"Read the code structured into a tree. "
"If you set specific line, it will show the code from the specific line. "
"Input should be filename, depth, and specific line if you want. ex. test.py|2 or test.py|3|print('hello world') "
"and the output will be list of (line number: code). ",
)
def summary(self, inputs: str) -> str:
try:
output = CodeReader.summary(inputs)
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.SUMMARY, Input Commands: {inputs} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.APPEND",
description="Append code to the existing file. "
"If the code is completed, use the Terminal tool to execute it, if not, append the code through the this tool. "
"Input should be filename and code to append. "
"Input code must be the code that should be appended, NOT whole code. "
"ex. test.py\nprint('hello world')\n "
"and the output will be last 3 lines.",
)
def append(self, inputs: str) -> str:
try:
code = CodeWriter.append(inputs)
output = "Last 3 line was:\n" + "\n".join(code.split("\n")[-3:])
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.APPEND, Input: {inputs} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.WRITE",
description="Write code to create a new tool. "
"If the code is completed, use the Terminal tool to execute it, if not, append the code through the CodeEditor.APPEND tool. "
"Input should be formatted like: "
"<filename>\n<code>\n\n"
"Here is an example: "
"test.py\nmessage = 'hello world'\nprint(message)\n"
"\n"
"The output will be last 3 lines you wrote.",
)
def write(self, inputs: str) -> str:
try:
code = CodeWriter.write(inputs.lstrip())
output = "Last 3 line was:\n" + "\n".join(code.split("\n")[-3:])
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.WRITE, Input: {inputs} " f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.PATCH",
description="Patch the code to correct the error if an error occurs or to improve it. "
"Input is a list of patches. The patch is separated by {seperator}. ".format(
seperator=CodePatcher.separator.replace("\n", "\\n")
)
+ "Each patch has to be formatted like below.\n"
"<filepath>|<start_line>,<start_col>|<end_line>,<end_col>|<new_code>"
"Here is an example. If the original code is:\n"
"print('hello world')\n"
"and you want to change it to:\n"
"print('hi corca')\n"
"then the patch should be:\n"
"test.py|1,8|1,19|hi corca\n"
"Code between start and end will be replaced with new_code. "
"The output will be written/deleted bytes or error message. ",
)
def patch(self, patches: str) -> str:
try:
w, d = CodePatcher.patch(patches)
output = f"successfully wrote {w}, deleted {d}"
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.PATCH, Input Patch: {patches} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.DELETE",
description="Delete code in file for a new start. "
"Input should be filename."
"ex. test.py "
"Output will be success or error message.",
)
def delete(self, inputs: str, filepath: str) -> str:
try:
with open(filepath, "w") as f:
f.write("")
output = "success"
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.DELETE, Input filename: {inputs} "
f"Output Answer: {output}"
)
return output
# ---------------- end
@tool(
name="CodeEditor.READ",
description="Read and understand code. "
"Input should be filename and line number group. ex. test.py|1-10 "
"and the output will be code. ",
)
def code_editor_read(self, inputs: str) -> str:
try:
output = CodeReader.read(inputs)
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.READ, Input Commands: {inputs} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.SUMMARY",
description="Summary code. "
"Read the code structured into a tree. "
"If you set specific line, it will show the code from the specific line. "
"Input should be filename, depth, and specific line if you want. ex. test.py|2 or test.py|3|print('hello world') "
"and the output will be list of (line number: code). ",
)
def code_editor_summary(self, inputs: str) -> str:
try:
output = CodeReader.summary(inputs)
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.SUMMARY, Input Commands: {inputs} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.APPEND",
description="Append code to the existing file. "
"If the code is completed, use the Terminal tool to execute it, if not, append the code through the this tool. "
"Input should be filename and code to append. "
"Input code must be the code that should be appended, NOT whole code. "
"ex. test.py\nprint('hello world')\n "
"and the output will be last 3 lines.",
)
def code_editor_append(self, inputs: str) -> str:
try:
code = CodeWriter.append(inputs)
output = "Last 3 line was:\n" + "\n".join(code.split("\n")[-3:])
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.APPEND, Input: {inputs} " f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.WRITE",
description="Write code to create a new tool. "
"If the code is completed, use the Terminal tool to execute it, if not, append the code through the CodeEditor.APPEND tool. "
"Input should be formatted like: "
"<filename>\n<code>\n\n"
"Here is an example: "
"test.py\nmessage = 'hello world'\nprint(message)\n"
"\n"
"The output will be last 3 lines you wrote.",
)
def code_editor_write(self, inputs: str) -> str:
try:
code = CodeWriter.write(inputs.lstrip())
output = "Last 3 line was:\n" + "\n".join(code.split("\n")[-3:])
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.WRITE, Input: {inputs} " f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.PATCH",
description="Patch the code to correct the error if an error occurs or to improve it. "
"Input is a list of patches. The patch is separated by {seperator}. ".format(
seperator=CodePatcher.separator.replace("\n", "\\n")
)
+ "Each patch has to be formatted like below.\n"
"<filepath>|<start_line>,<start_col>|<end_line>,<end_col>|<new_code>"
"Here is an example. If the original code is:\n"
"print('hello world')\n"
"and you want to change it to:\n"
"print('hi corca')\n"
"then the patch should be:\n"
"test.py|1,8|1,19|hi corca\n"
"Code between start and end will be replaced with new_code. "
"The output will be written/deleted bytes or error message. ",
)
def code_editor_patch(self, patches: str) -> str:
try:
w, d = CodePatcher.patch(patches)
output = f"successfully wrote {w}, deleted {d}"
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.PATCH, Input Patch: {patches} "
f"Output Answer: {output}"
)
return output
@tool(
name="CodeEditor.DELETE",
description="Delete code in file for a new start. "
"Input should be filename."
"ex. test.py "
"Output will be success or error message.",
)
def code_editor_delete(self, inputs: str, filepath: str) -> str:
try:
with open(filepath, "w") as f:
f.write("")
output = "success"
except Exception as e:
output = str(e)
logger.debug(
f"\nProcessed CodeEditor.DELETE, Input filename: {inputs} "
f"Output Answer: {output}"
)
return output

@ -1,17 +0,0 @@
from langchain.agents.agent_toolkits import FileManagementToolkit
from tempfile import TemporaryDirectory
# We'll make a temporary directory to avoid clutter
working_directory = TemporaryDirectory()
toolkit = FileManagementToolkit(
root_dir=str(working_directory.name)
) # If you don't provide a root_dir, operations will default to the current working directory
toolkit.get_tools()
file_management_tools = FileManagementToolkit(
root_dir=str(working_directory.name),
selected_tools=["read_file", "write_file", "list_directory"],
).get_tools()
read_tool, write_tool, list_tool = file_management_tools
Loading…
Cancel
Save