Introduction
In this tutorial, you'll learn how to build and use a SuperAgent framework inspired by ByteDance's DeerFlow 2.0. This framework orchestrates sub-agents, memory systems, and sandboxes to execute complex tasks autonomously. While the original DeerFlow is a sophisticated system, we'll create a simplified yet functional prototype that demonstrates core concepts.
By the end of this tutorial, you'll have a working SuperAgent that can plan, execute, and remember task outcomes - a crucial step toward building autonomous AI systems.
Prerequisites
- Python 3.8 or higher
- Basic understanding of Python classes and object-oriented programming
- Knowledge of AI/ML concepts (agents, planning, memory)
- Installed packages:
openai,langchain,pydantic
Step-by-Step Instructions
1. Set Up Your Development Environment
First, create a new Python project directory and install the required dependencies:
mkdir deerflow-superagent
cd deerflow-superagent
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
pip install openai langchain pydantic
Why: We need these libraries to interact with language models, build agent components, and structure our data.
2. Create the Base Agent Class
Create a file called base_agent.py to define the foundational agent structure:
from pydantic import BaseModel
from typing import List, Dict, Any
class AgentConfig(BaseModel):
name: str
description: str
capabilities: List[str]
class BaseAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.memory = {}
def execute(self, task: str) -> Dict[str, Any]:
raise NotImplementedError("Subclasses must implement execute method")
def remember(self, key: str, value: Any):
self.memory[key] = value
def recall(self, key: str) -> Any:
return self.memory.get(key)
Why: This creates a reusable base class that all sub-agents will inherit from, ensuring consistency in structure and memory management.
3. Implement a Planning Sub-Agent
Create planning_agent.py to handle task decomposition:
from base_agent import BaseAgent, AgentConfig
from typing import List, Dict, Any
import openai
class PlanningAgent(BaseAgent):
def __init__(self, config: AgentConfig, api_key: str):
super().__init__(config)
openai.api_key = api_key
def execute(self, task: str) -> Dict[str, Any]:
prompt = f"Break down this task into 2-3 actionable steps: {task}"
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
max_tokens=150
)
steps = response.choices[0].message.content.split('\n')
self.remember('planned_steps', steps)
return {
"status": "planned",
"steps": steps,
"task": task
}
Why: The planning agent breaks complex tasks into manageable sub-tasks, which is essential for executing multi-step processes.
4. Create a Task Execution Agent
Implement execution_agent.py to carry out specific steps:
from base_agent import BaseAgent, AgentConfig
from typing import List, Dict, Any
import openai
class ExecutionAgent(BaseAgent):
def __init__(self, config: AgentConfig, api_key: str):
super().__init__(config)
openai.api_key = api_key
def execute(self, step: str) -> Dict[str, Any]:
prompt = f"Execute this step and provide a clear result: {step}"
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
max_tokens=200
)
result = response.choices[0].message.content
self.remember('last_execution', result)
return {
"status": "executed",
"step": step,
"result": result
}
Why: This agent performs the actual work, using LLMs to execute specific tasks and return results.
5. Build the SuperAgent Orchestrator
Create superagent.py to coordinate all components:
from base_agent import BaseAgent
from planning_agent import PlanningAgent
from execution_agent import ExecutionAgent
from typing import List, Dict, Any
class SuperAgent:
def __init__(self, api_key: str):
self.api_key = api_key
self.planning_agent = PlanningAgent(
AgentConfig(
name="Planner",
description="Breaks down tasks into steps",
capabilities=["planning"]
),
api_key
)
self.execution_agent = ExecutionAgent(
AgentConfig(
name="Executor",
description="Executes individual steps",
capabilities=["execution"]
),
api_key
)
def execute_task(self, task: str) -> Dict[str, Any]:
# Step 1: Plan
plan_result = self.planning_agent.execute(task)
# Step 2: Execute each step
results = []
for step in plan_result["steps"]:
execution_result = self.execution_agent.execute(step)
results.append(execution_result)
# Step 3: Compile final result
final_result = {
"task": task,
"plan": plan_result["steps"],
"executions": results,
"summary": f"Task completed with {len(results)} steps"
}
return final_result
Why: The SuperAgent acts as the coordinator, managing the flow between planning and execution agents, and collecting results.
6. Test Your SuperAgent
Create main.py to run a sample task:
from superagent import SuperAgent
def main():
# Initialize SuperAgent with your OpenAI API key
agent = SuperAgent(api_key="your-openai-api-key")
# Execute a complex task
task = "Research the impact of AI on education and write a summary"
result = agent.execute_task(task)
print("Task Result:")
print(f"Original Task: {result['task']}")
print(f"Plan: {result['plan']}")
print(f"Summary: {result['summary']}")
for i, execution in enumerate(result['executions']):
print(f"Step {i+1}: {execution['step']}")
print(f"Result: {execution['result'][:100]}...")
n
if __name__ == "__main__":
main()
Why: This test demonstrates the complete workflow of your SuperAgent framework from planning to execution.
7. Add Memory Management
Enhance base_agent.py to include a more sophisticated memory system:
from pydantic import BaseModel
from typing import List, Dict, Any
import json
class MemoryManager:
def __init__(self):
self.memory = {}
def store(self, key: str, data: Any):
self.memory[key] = {
"data": data,
"timestamp": str(datetime.now())
}
def retrieve(self, key: str) -> Any:
entry = self.memory.get(key)
return entry["data"] if entry else None
class BaseAgent:
def __init__(self, config: AgentConfig):
self.config = config
self.memory = MemoryManager()
def execute(self, task: str) -> Dict[str, Any]:
raise NotImplementedError("Subclasses must implement execute method")
Why: A proper memory system allows agents to remember previous interactions and build on past knowledge.
Summary
You've now built a simplified yet functional SuperAgent framework inspired by ByteDance's DeerFlow 2.0. This system demonstrates core concepts of autonomous AI agents: planning, execution, and memory management. While this prototype is basic, it provides a foundation for building more sophisticated systems that can tackle increasingly complex tasks.
Key takeaways include understanding how to structure agents, orchestrate their interactions, and manage shared memory - all essential components for creating truly autonomous AI systems.
