An innovative system designed for complex task automation through a sophisticated graph-based architecture. PlanAI integrates traditional computations and cutting-edge AI technologies to enable versatile and efficient workflow management.
Get StartedConstruct dynamic workflows comprising interconnected TaskWorkers for highly customizable automation.
Combine conventional computations (e.g., API calls) with powerful LLM-driven operations, leveraging Retrieval-Augmented Generation (RAG) capabilities.
Ensure data integrity and type consistency across workflows with Pydantic-validated input and output.
Utilize type-aware routing to efficiently manage data flow between nodes, adapting to multiple downstream consumers.
Trace the lineage and origin of each Task as it flows through the workflow, enabling detailed analysis and debugging of complex processes.
Improve LLMTaskWorker prompts automatically using AI-driven optimization, enhancing performance with minimal manual intervention.
PlanAI has been used to create a sophisticated system for generating high-quality question and answer pairs from textbook content. This example demonstrates PlanAI's capability to manage complex, multi-step workflows involving AI-powered text processing and content generation.
You can install PlanAI using pip:
pip install planai
Here's a simple example of how to use PlanAI:
from planai import Graph, TaskWorker, LLMTaskWorker, llm_from_config
# Define custom TaskWorkers
class CustomDataProcessor(TaskWorker):
output_types = [ProcessedData]
def consume_work(self, task: RawData):
processed_data = self.process(task.data)
self.publish_work(ProcessedData(data=processed_data))
# Define an LLM-powered task
class AIAnalyzer(LLMTaskWorker):
prompt = "Analyze the provided data and derive insights"
output_types = [AnalysisResult]
def consume_work(self, task: ProcessedData):
super().consume_work(task)
# Create and run the workflow
graph = Graph(name="Data Analysis Workflow")
data_processor = CustomDataProcessor()
ai_analyzer = AIAnalyzer(
llm=llm_from_config(provider="openai", model_name="gpt-4o")
)
graph.add_workers(data_processor, ai_analyzer)
graph.set_dependency(data_processor, ai_analyzer)
initial_data = RawData(data="Some raw data")
graph.run(initial_tasks=[(data_processor, initial_data)])
For more detailed examples and advanced usage, please refer to our documentation.
Join the growing community of PlanAI users and take your productivity to the next level.