# Dria-SDK
Dria SDK is a powerful SDK for building and executing AI-powered workflows and pipelines. It provides a flexible and extensible framework for creating complex AI tasks, managing distributed computing resources, and handling various AI models.
## Table of Contents
1. [Installation](#installation)
2. [Features](#features)
3. [Login](#login)
4. [Getting Started](#getting-started)
5. [Usage Examples](#usage-examples)
6. [API Usage](#api-usage)
7. [License](#license)
## Installation
To install Dria SDK, you can use pip:
```bash
pip install dria
```
## Features
- Create and manage AI workflows and pipelines
- Support for multiple AI models
- Distributed task execution
- Flexible configuration options
- Built-in error handling and retries
- Extensible callback system
## Login
Dria SDK uses authentication token for sending tasks to the Dria Network. You should get your rpc token from [Dria Login API](https://dkn.dria.co/auth/generate-token).
## Getting Started
To get started with Dria SDK, you'll need to set up your environment and initialize the Dria client:
```python
import os
from dria.client import Dria
# Initialize the Dria client
dria = Dria(rpc_token=os.environ["DRIA_RPC_TOKEN"])
# Initialize the client (should be called before using any other methods)
await dria.initialize()
```
## Usage Examples
### Creating a Simple Workflow
Here's an example of creating a simple workflow for generating a poem:
```python
import asyncio
from dria.client import Dria
from dria.models import Task, TaskResult
from dria.models.enums import Model
from dria.workflows.lib.poem_generator import poem
dria = Dria()
async def generate_poem(prompt: str) -> list[TaskResult]:
task = Task(
workflow=poem(prompt),
models=[Model.QWEN2_5_7B_FP16]
)
await dria.push(task)
return await dria.fetch(task_id=task.id)
async def main():
await dria.initialize()
result = await generate_poem("Write a poem about love")
if __name__ == "__main__":
asyncio.run(main())
```
### Building a Complex Pipeline
For more complex scenarios, you can use the `PipelineBuilder` to create multi-step pipelines:
```python
from dria.client import Dria
from dria.models import Model, TaskInput
from dria.pipelines import PipelineConfig, StepConfig, PipelineBuilder, StepBuilder
from workflows import generate_entries, generate_subtopics
async def create_subtopic_pipeline(dria: Dria, topic, config: PipelineConfig = PipelineConfig(), max_depth=1):
pipeline = PipelineBuilder(config, dria)
depth = 0
# handles single topic output
subtopics = StepBuilder(input=TaskInput(topics=[topic]), config=StepConfig(models=[Model.QWEN2_5_7B_FP16,
Model.GPT4O]),
workflow=generate_subtopics).broadcast().build()
pipeline.add_step(subtopics)
while depth < max_depth:
# handles multiple topics
subtopics = StepBuilder(workflow=generate_subtopics,
config=StepConfig(models=[Model.QWEN2_5_7B_FP16, Model.GPT4O])).scatter().build()
pipeline.add_step(subtopics)
depth += 1
# entry generation
entries = StepBuilder(workflow=generate_entries, config=StepConfig(min_compute=0.8)).build()
pipeline.add_step(entries)
return pipeline.build()
```
### API Usage
You can use the Dria SDK on the API level to create your own workflows and pipelines.
```python
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field
from dria.client import Dria
from dria.pipelines.pipeline import PipelineConfig, Pipeline
from pipeline import create_subtopic_pipeline
app = FastAPI(title="Dria SDK Example")
dria = Dria()
@app.on_event("startup")
async def startup_event():
await dria.initialize()
class PipelineRequest(BaseModel):
input_text: str = Field(..., description="The input text for the pipeline to process")
class PipelineResponse(BaseModel):
pipeline_id: str = Field(..., description="Unique identifier for the created pipeline")
pipeline_config = PipelineConfig(retry_interval=5)
pipelines = {}
@app.post("/run_pipeline", response_model=PipelineResponse)
async def run_pipeline(request: PipelineRequest, background_tasks: BackgroundTasks):
pipeline = await create_subtopic_pipeline(dria, request.input_text, pipeline_config)
pipelines[pipeline.pipeline_id] = pipeline
background_tasks.add_task(pipeline.execute)
return PipelineResponse(pipeline_id=pipeline.pipeline_id)
@app.get("/pipeline_status/{pipeline_id}")
async def get_pipeline_status(pipeline_id: str):
if pipeline_id not in pipelines:
raise HTTPException(status_code=404, detail="Pipeline not found")
pipeline = pipelines[pipeline_id]
state, status, result = pipeline.poll()
if result is not None:
del pipelines[pipeline_id]
return {"status": status, "state": state, "result": result}
# Usage example:
# uvicorn main:app --host 0.0.0.0 --port 8005
```
For more detailed API documentation, see on our [documentation site](https://docs.dria.ai).
## License
Dria SDK is released under the [MIT License](https://opensource.org/licenses/MIT).
Raw data
{
"_id": null,
"home_page": "https://github.com/firstbatchxyz/dria-sdk",
"name": "dria",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.10",
"maintainer_email": null,
"keywords": "dria, decentralized-ai, ai, agentic",
"author": "Omer",
"author_email": "omer@firstbatch.xyz",
"download_url": "https://files.pythonhosted.org/packages/7f/e0/f5777602792cdbf486709bddc242481f5b2b9a9958c0d6cea879032dd346/dria-0.0.17.tar.gz",
"platform": null,
"description": "# Dria-SDK\n\nDria SDK is a powerful SDK for building and executing AI-powered workflows and pipelines. It provides a flexible and extensible framework for creating complex AI tasks, managing distributed computing resources, and handling various AI models.\n\n## Table of Contents\n\n1. [Installation](#installation)\n2. [Features](#features)\n3. [Login](#login)\n4. [Getting Started](#getting-started)\n5. [Usage Examples](#usage-examples)\n6. [API Usage](#api-usage)\n7. [License](#license)\n\n## Installation\n\nTo install Dria SDK, you can use pip:\n\n```bash\npip install dria\n```\n\n## Features\n\n- Create and manage AI workflows and pipelines\n- Support for multiple AI models\n- Distributed task execution\n- Flexible configuration options\n- Built-in error handling and retries\n- Extensible callback system\n\n## Login\n\nDria SDK uses authentication token for sending tasks to the Dria Network. You should get your rpc token from [Dria Login API](https://dkn.dria.co/auth/generate-token).\n\n## Getting Started\n\nTo get started with Dria SDK, you'll need to set up your environment and initialize the Dria client:\n\n```python\nimport os\nfrom dria.client import Dria\n\n# Initialize the Dria client\ndria = Dria(rpc_token=os.environ[\"DRIA_RPC_TOKEN\"])\n\n# Initialize the client (should be called before using any other methods)\nawait dria.initialize()\n```\n\n\n\n## Usage Examples\n\n### Creating a Simple Workflow\n\nHere's an example of creating a simple workflow for generating a poem:\n\n```python\nimport asyncio\n\nfrom dria.client import Dria\nfrom dria.models import Task, TaskResult\nfrom dria.models.enums import Model\nfrom dria.workflows.lib.poem_generator import poem\n\ndria = Dria()\n\nasync def generate_poem(prompt: str) -> list[TaskResult]:\n task = Task(\n workflow=poem(prompt),\n models=[Model.QWEN2_5_7B_FP16]\n )\n await dria.push(task)\n return await dria.fetch(task_id=task.id)\n\nasync def main():\n \n await dria.initialize()\n result = await generate_poem(\"Write a poem about love\")\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### Building a Complex Pipeline\n\nFor more complex scenarios, you can use the `PipelineBuilder` to create multi-step pipelines:\n\n```python\nfrom dria.client import Dria\nfrom dria.models import Model, TaskInput\nfrom dria.pipelines import PipelineConfig, StepConfig, PipelineBuilder, StepBuilder\nfrom workflows import generate_entries, generate_subtopics\n\n\nasync def create_subtopic_pipeline(dria: Dria, topic, config: PipelineConfig = PipelineConfig(), max_depth=1):\n pipeline = PipelineBuilder(config, dria)\n depth = 0\n\n # handles single topic output\n subtopics = StepBuilder(input=TaskInput(topics=[topic]), config=StepConfig(models=[Model.QWEN2_5_7B_FP16,\n Model.GPT4O]),\n workflow=generate_subtopics).broadcast().build()\n pipeline.add_step(subtopics)\n\n while depth < max_depth:\n # handles multiple topics\n subtopics = StepBuilder(workflow=generate_subtopics,\n config=StepConfig(models=[Model.QWEN2_5_7B_FP16, Model.GPT4O])).scatter().build()\n pipeline.add_step(subtopics)\n depth += 1\n\n # entry generation\n entries = StepBuilder(workflow=generate_entries, config=StepConfig(min_compute=0.8)).build()\n pipeline.add_step(entries)\n return pipeline.build()\n\n```\n\n### API Usage\n\nYou can use the Dria SDK on the API level to create your own workflows and pipelines.\n\n```python\nfrom fastapi import FastAPI, HTTPException, BackgroundTasks\nfrom pydantic import BaseModel, Field\nfrom dria.client import Dria\nfrom dria.pipelines.pipeline import PipelineConfig, Pipeline\nfrom pipeline import create_subtopic_pipeline\n\napp = FastAPI(title=\"Dria SDK Example\")\ndria = Dria()\n\n@app.on_event(\"startup\")\nasync def startup_event():\n await dria.initialize()\n\nclass PipelineRequest(BaseModel):\n input_text: str = Field(..., description=\"The input text for the pipeline to process\")\n\nclass PipelineResponse(BaseModel):\n pipeline_id: str = Field(..., description=\"Unique identifier for the created pipeline\")\n\npipeline_config = PipelineConfig(retry_interval=5)\npipelines = {}\n\n@app.post(\"/run_pipeline\", response_model=PipelineResponse)\nasync def run_pipeline(request: PipelineRequest, background_tasks: BackgroundTasks):\n pipeline = await create_subtopic_pipeline(dria, request.input_text, pipeline_config)\n pipelines[pipeline.pipeline_id] = pipeline\n background_tasks.add_task(pipeline.execute)\n return PipelineResponse(pipeline_id=pipeline.pipeline_id)\n\n@app.get(\"/pipeline_status/{pipeline_id}\")\nasync def get_pipeline_status(pipeline_id: str):\n if pipeline_id not in pipelines:\n raise HTTPException(status_code=404, detail=\"Pipeline not found\")\n \n pipeline = pipelines[pipeline_id]\n state, status, result = pipeline.poll()\n \n if result is not None:\n del pipelines[pipeline_id]\n \n return {\"status\": status, \"state\": state, \"result\": result}\n\n# Usage example:\n# uvicorn main:app --host 0.0.0.0 --port 8005\n\n```\n\nFor more detailed API documentation, see on our [documentation site](https://docs.dria.ai).\n\n## License\n\nDria SDK is released under the [MIT License](https://opensource.org/licenses/MIT).\n",
"bugtrack_url": null,
"license": null,
"summary": "Dria SDK - A Python library for interacting with the Dria AI Network",
"version": "0.0.17",
"project_urls": {
"Bug Tracker": "https://github.com/firstbatchxyz/dria-sdk/issues",
"Documentation": "https://github.com/firstbatchxyz/dria-sdk",
"Homepage": "https://github.com/firstbatchxyz/dria-sdk",
"Repository": "https://github.com/firstbatchxyz/dria-sdk"
},
"split_keywords": [
"dria",
" decentralized-ai",
" ai",
" agentic"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "fb45b6c064bca0ad70c5d7a9349a1be7d902926ec7396bb5436bfd50796ba121",
"md5": "695ba70178da787befbeb0d1dbef4495",
"sha256": "f6632f75806dc16fd3e5b086694e74ebd3ede8561567a6ca4aef5d101d1bb2e6"
},
"downloads": -1,
"filename": "dria-0.0.17-py3-none-any.whl",
"has_sig": false,
"md5_digest": "695ba70178da787befbeb0d1dbef4495",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 33816,
"upload_time": "2024-10-07T09:54:18",
"upload_time_iso_8601": "2024-10-07T09:54:18.073815Z",
"url": "https://files.pythonhosted.org/packages/fb/45/b6c064bca0ad70c5d7a9349a1be7d902926ec7396bb5436bfd50796ba121/dria-0.0.17-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "7fe0f5777602792cdbf486709bddc242481f5b2b9a9958c0d6cea879032dd346",
"md5": "1b2db09ae256a77366f3a3d8c705e642",
"sha256": "ff78c7f6662b68c904113ab4b7e69ef0af04668fec19cfbe9dec3e747de695fc"
},
"downloads": -1,
"filename": "dria-0.0.17.tar.gz",
"has_sig": false,
"md5_digest": "1b2db09ae256a77366f3a3d8c705e642",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 27829,
"upload_time": "2024-10-07T09:54:19",
"upload_time_iso_8601": "2024-10-07T09:54:19.770071Z",
"url": "https://files.pythonhosted.org/packages/7f/e0/f5777602792cdbf486709bddc242481f5b2b9a9958c0d6cea879032dd346/dria-0.0.17.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-10-07 09:54:19",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "firstbatchxyz",
"github_project": "dria-sdk",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "dria"
}
JSON
