# 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.pipeline 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.pipeline.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 pipelines to process")
class PipelineResponse(BaseModel):
pipeline_id: str = Field(..., description="Unique identifier for the created pipelines")
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": "andthattoo",
"author_email": "omer@firstbatch.xyz",
"download_url": "https://files.pythonhosted.org/packages/0c/3d/2d323c0330692da767c89dbcfdc33f1444d708be41fe7bad84c34df36da8/dria-0.0.45.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.pipeline 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.pipeline.pipeline import PipelineConfig, Pipeline\nfrom pipeline import create_subtopic_pipeline\n\napp = FastAPI(title=\"Dria SDK Example\")\ndria = Dria()\n\n\n@app.on_event(\"startup\")\nasync def startup_event():\n await dria.initialize()\n\n\nclass PipelineRequest(BaseModel):\n input_text: str = Field(..., description=\"The input text for the pipelines to process\")\n\n\nclass PipelineResponse(BaseModel):\n pipeline_id: str = Field(..., description=\"Unique identifier for the created pipelines\")\n\n\npipeline_config = PipelineConfig(retry_interval=5)\npipelines = {}\n\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\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.45",
"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": "448cfe9d74b07da1216770aa630c76de987b64d8f45952c5c1f5c78eb9fd561f",
"md5": "9b3cf4ce47e69b4601fc96b762a34d21",
"sha256": "359c3d6aa11dcc68d024095c1f02d1d9ff9b64b429ad4c171bac5ba50e1e8d48"
},
"downloads": -1,
"filename": "dria-0.0.45-py3-none-any.whl",
"has_sig": false,
"md5_digest": "9b3cf4ce47e69b4601fc96b762a34d21",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 118243,
"upload_time": "2024-10-23T11:02:42",
"upload_time_iso_8601": "2024-10-23T11:02:42.998101Z",
"url": "https://files.pythonhosted.org/packages/44/8c/fe9d74b07da1216770aa630c76de987b64d8f45952c5c1f5c78eb9fd561f/dria-0.0.45-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "0c3d2d323c0330692da767c89dbcfdc33f1444d708be41fe7bad84c34df36da8",
"md5": "cf6a7aa65bbdc37f48d980a323e6c079",
"sha256": "52d41cd0e79f7ce2eca765509f019994f203c61818e2cab3675102b242692314"
},
"downloads": -1,
"filename": "dria-0.0.45.tar.gz",
"has_sig": false,
"md5_digest": "cf6a7aa65bbdc37f48d980a323e6c079",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 72008,
"upload_time": "2024-10-23T11:02:44",
"upload_time_iso_8601": "2024-10-23T11:02:44.873950Z",
"url": "https://files.pythonhosted.org/packages/0c/3d/2d323c0330692da767c89dbcfdc33f1444d708be41fe7bad84c34df36da8/dria-0.0.45.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-10-23 11:02:44",
"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
