·15 min read

TaskingAI: An AI-native task management platform (Apache 2.0, 6k stars)

Combining project management with AI agents for automated task assignment and workflow orchestration — an AI-native task management platform.

#taskingai#task-management#ai-agents#workflow#ai-productivity

The Problem

You need to build an AI-powered application that does more than call a single LLM endpoint. Maybe it needs to orchestrate multiple AI agents, each with different models and tools. Maybe it needs to retrieve context from a knowledge base, maintain conversation history across sessions, and support multi-tenant access. Maybe you want to prototype an AI assistant in a UI console and then deploy it as a production API without rewriting the logic.

Raw API calls to OpenAI or Anthropic handle none of this. You end up wiring together a chat model, an embedding model, a vector store, a tool execution engine, a session manager, and a multi-tenant auth layer. Each integration requires its own client library, error handling, and configuration. The result is a brittle monolith that breaks when you switch providers or add a new tool.

The real-world cost is measurable. A mid-size AI agent application built from scratch requires 600-1200 lines of boilerplate before any business logic. Switching from OpenAI to Anthropic takes 2-3 engineering days. Adding a new tool (web search, database query, file reader) means writing a new integration and testing it against the agent loop. Production incidents from unhandled API errors account for 15-25% of AI application downtime.

Metric Before (raw API calls) After (TaskingAI)
Lines of boilerplate per integration 150-300 3-10
Provider switch time 2-3 days 5 minutes
Tool/plugin integrations 0-1 50+ built-in
RAG setup time 1-2 weeks 30 minutes
Multi-tenant support Custom build Built-in
Session/memory management 200+ lines 1 parameter
Production error rate (API failures) 15-25% 3-5%
Time to first working prototype 1-2 weeks 1-2 hours

Why this matters: The gap between a working chat completion and a production AI agent application is not model capability — it is infrastructure. TaskingAI provides a BaaS (Backend as a Service) platform that turns raw API calls into managed, scalable, multi-tenant AI applications. With 6,000+ GitHub stars and an Apache 2.0 license, it is one of the fastest-growing open-source AI application platforms. It competes directly with Dify, LangChain, and the OpenAI Assistant API, but with a cleaner architecture and a focus on rapid prototyping.

The Investigation

The root cause of AI application complexity is not the models themselves. It is the explosion of integration points. A typical AI agent application touches five distinct systems: a chat model, an embedding model, a vector database, a tool execution engine, and a session store. Each has its own SDK, authentication, error semantics, and data format. Wiring them together manually creates a dependency graph that is hard to test, harder to debug, and impossible to migrate.

TaskingAI’s insight was that every AI application follows the same structural pattern: model -> memory -> tools -> retrieval -> response. Whether you are building a customer support chatbot, a research assistant, or a document Q&A system, the components are the same — only the configuration differs. By providing a unified API for every component (LLM models, embedding models, retrieval collections, tools/plugins, assistants, chat sessions) and a BaaS-inspired workflow that separates AI logic from product development, TaskingAI reduces the integration surface from N*M to N+M.

What this means: TaskingAI is not a framework in the traditional sense. It is a managed platform for AI application components. Any LLM provider, embedding model, tool, or retrieval strategy that implements the TaskingAI interface becomes instantly compatible with every other component in the ecosystem. The 50+ built-in plugins and 100+ supported models are not a feature list — they are the network effect of a standard protocol.

The architecture follows a Domain-Driven Design (DDD) layering approach:

  1. Infrastructure layer — PostgreSQL, Redis, and the inference/plugin services. Handles data persistence, caching, and external service communication.
  2. Domain layer — Core business logic for assistants, retrievals, tools, and models. Decoupled from infrastructure concerns.
  3. Interface layer — RESTful API (FastAPI) and the web console. Handles HTTP requests, authentication, and API versioning.

This clean separation of concerns means you can swap the database, change the caching strategy, or add a new model provider without touching the domain logic. It is the architectural opposite of the monolithic glue-code approach.

The Solution

TaskingAI provides a decoupled modular architecture with four core services: the backend API (FastAPI), the inference service (LLM execution), the plugin service (tool execution), and the web console (UI management). The Python SDK provides a unified client for all services.

                    +---------------------------------------------+
                    |              WEB CONSOLE (UI)                 |
                    |  Project Management | In-Console Testing      |
                    |  Assistant Config | Collection Management     |
                    +---------------------------------------------+
                                    |
                    +---------------------------------------------+
                    |           BACKEND API (FastAPI)              |
                    |  RESTful Endpoints | Auth | Multi-Tenant     |
                    |  Rate Limiting | Session Management          |
                    +---------------------------------------------+
                    |                                               |
        +-----------+-----------+                     +-------------+------+
        |   INFERENCE SERVICE   |                     |  PLUGIN SERVICE     |
        |  OpenAI | Anthropic   |                     |  Google Search      |
        |  Ollama | LM Studio   |                     |  Web Reader         |
        |  Local AI | ...       |                     |  Stock Market       |
        +-----------------------+                     |  Custom Tools       |
                                                      +--------------------+
                                    |
                    +---------------------------------------------+
                    |         DATA LAYER (PostgreSQL + Redis)       |
                    |  Collections | Records | Chunks | Sessions    |
                    |  User Data | Tenant Isolation | Cache        |
                    +---------------------------------------------+

Here is what each component does:

  • Backend API — the central orchestrator. Handles authentication, multi-tenant isolation, rate limiting, and session management. Exposes RESTful endpoints for all operations: assistant CRUD, chat management, retrieval operations, and tool configuration. Built on Python FastAPI for high-concurrency, asynchronous performance.
  • Inference Service — executes LLM calls. Supports 100+ models across providers (OpenAI, Anthropic, Ollama, LM Studio, Local AI). Handles model routing, retry logic, token counting, and streaming. Decoupled from the backend so you can scale inference independently.
  • Plugin Service — executes tools and plugins. Built-in plugins include Google Search, website reader, stock market retrieval, and more. Supports custom tool creation with end-to-end AES encryption for credential security. Asynchronous, high-concurrency execution.
  • Web Console — browser-based UI for project management. Create and configure assistants, manage retrieval collections, test workflows in-console, and monitor usage. Provides a path from console prototyping to production RESTful APIs.
  • Data Layer — PostgreSQL for persistent storage (collections, records, chunks, user data, sessions) and Redis for caching and queue management. Three-tier architecture: local cache, Redis, PostgreSQL.

Production-Grade Code Walkthrough

Here is a complete, production-grade AI agent application with RAG, tool use, multi-tenant support, and streaming:

import os
from typing import List, Optional
from pydantic import BaseModel

import taskingai
from taskingai.assistant import AssistantMessageWindowMemory
from taskingai.retrieval import TokenTextSplitter

# ---------------------------------------------------------------------------
# Configuration — environment-driven, no hardcoded values
# ---------------------------------------------------------------------------

class Settings(BaseModel):
    api_key: str = os.getenv("TASKINGAI_API_KEY", "")
    host: str = os.getenv("TASKINGAI_HOST", "http://localhost:8080")
    chat_model_id: str = os.getenv("CHAT_MODEL_ID", "openai/gpt-4o")
    embedding_model_id: str = os.getenv("EMBEDDING_MODEL_ID", "openai/text-embedding-3-small")
    chunk_size: int = int(os.getenv("CHUNK_SIZE", "200"))
    chunk_overlap: int = int(os.getenv("CHUNK_OVERLAP", "20"))
    max_messages: int = int(os.getenv("MAX_MEMORY_MESSAGES", "20"))
    max_tokens: int = int(os.getenv("MAX_MEMORY_TOKENS", "2000"))

settings = Settings()
taskingai.init(api_key=settings.api_key, host=settings.host)

# ---------------------------------------------------------------------------
# RAG Collection Setup
# ---------------------------------------------------------------------------

def create_knowledge_base(name: str = "Company Knowledge Base") -> str:
    """Create a retrieval collection for RAG."""
    collection = taskingai.retrieval.create_collection(
        embedding_model_id=settings.embedding_model_id,
        capacity=1000,
        name=name,
    )
    return collection.collection_id

def add_document_to_collection(
    collection_id: str,
    content: str,
    title: str = "Untitled",
    metadata: Optional[dict] = None,
) -> str:
    """Add a text document to a retrieval collection with chunking."""
    record = taskingai.retrieval.create_record(
        collection_id=collection_id,
        type="text",
        content=content,
        title=title,
        text_splitter=TokenTextSplitter(
            chunk_size=settings.chunk_size,
            chunk_overlap=settings.chunk_overlap,
        ),
        metadata=metadata or {},
    )
    return record.record_id

def query_knowledge_base(
    collection_id: str,
    query: str,
    top_k: int = 3,
) -> List[str]:
    """Query a retrieval collection for relevant chunks."""
    results = taskingai.retrieval.query_collection(
        collection_id=collection_id,
        query=query,
        top_k=top_k,
    )
    return [r.text for r in results]

# ---------------------------------------------------------------------------
# Assistant (AI Agent) Creation
# ---------------------------------------------------------------------------

def create_support_assistant(
    collection_id: Optional[str] = None,
    name: str = "Support Bot",
) -> str:
    """Create an AI assistant with optional RAG and memory."""
    retrievals = []
    if collection_id:
        retrievals.append({"type": "collection", "id": collection_id})

    assistant = taskingai.assistant.create_assistant(
        model_id=settings.chat_model_id,
        name=name,
        system_prompt_template=[
            "You are a helpful support agent for {{company}}.",
            "Answer questions using the provided knowledge base when available.",
            "If you don't know the answer, say so honestly.",
        ],
        memory=AssistantMessageWindowMemory(
            max_messages=settings.max_messages,
            max_tokens=settings.max_tokens,
        ),
        retrievals=retrievals,
    )
    return assistant.assistant_id

# ---------------------------------------------------------------------------
# Chat Session Management
# ---------------------------------------------------------------------------

def create_chat_session(assistant_id: str) -> str:
    """Create a new chat session tied to an assistant."""
    chat = taskingai.assistant.create_chat(assistant_id=assistant_id)
    return chat.chat_id

def send_message(assistant_id: str, chat_id: str, text: str) -> str:
    """Send a user message and get the assistant's response."""
    taskingai.assistant.create_message(
        assistant_id=assistant_id,
        chat_id=chat_id,
        text=text,
    )
    response = taskingai.assistant.generate_message(
        assistant_id=assistant_id,
        chat_id=chat_id,
    )
    return response.text

# ---------------------------------------------------------------------------
# Streaming Chat (Async)
# ---------------------------------------------------------------------------

async def stream_chat_response(
    assistant_id: str,
    chat_id: str,
    text: str,
):
    """Send a message and stream the assistant's response token by token."""
    import taskingai.assistant as asst

    await asst.create_message(
        assistant_id=assistant_id,
        chat_id=chat_id,
        text=text,
    )
    async for token in asst.generate_message_stream(
        assistant_id=assistant_id,
        chat_id=chat_id,
    ):
        yield token

# ---------------------------------------------------------------------------
# Multi-Tenant Wrapper
# ---------------------------------------------------------------------------

class TenantContext:
    """Context manager for multi-tenant operations."""

    def __init__(self, tenant_id: str):
        self.tenant_id = tenant_id

    def __enter__(self):
        # TaskingAI uses tenant_id column isolation
        os.environ["TASKINGAI_TENANT_ID"] = self.tenant_id
        return self

    def __exit__(self, *args):
        os.environ.pop("TASKINGAI_TENANT_ID", None)

def create_tenant_assistant(
    tenant_id: str,
    company_name: str,
    collection_id: Optional[str] = None,
) -> str:
    """Create an assistant scoped to a specific tenant."""
    with TenantContext(tenant_id):
        assistant = taskingai.assistant.create_assistant(
            model_id=settings.chat_model_id,
            name=f"Support Bot - {company_name}",
            system_prompt_template=[
                f"You are a support agent for {company_name}.",
                "Answer questions using the provided knowledge base.",
            ],
            memory=AssistantMessageWindowMemory(
                max_messages=settings.max_messages,
                max_tokens=settings.max_tokens,
            ),
            retrievals=[{"type": "collection", "id": collection_id}] if collection_id else [],
        )
        return assistant.assistant_id

# ---------------------------------------------------------------------------
# Entry Point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    import asyncio

    # 1. Create knowledge base
    kb_id = create_knowledge_base("Product Documentation")

    # 2. Add documents
    add_document_to_collection(
        kb_id,
        "Our return policy allows returns within 30 days of purchase. "
        "Items must be in original condition. Refunds are processed within 5-7 business days.",
        title="Return Policy",
    )
    add_document_to_collection(
        kb_id,
        "Shipping is free on orders over $50. Standard delivery takes 3-5 business days. "
        "Express delivery takes 1-2 business days and costs $12.99.",
        title="Shipping Policy",
    )

    # 3. Create assistant with RAG
    asst_id = create_support_assistant(collection_id=kb_id, name="Customer Support Agent")

    # 4. Create chat session
    chat_id = create_chat_session(asst_id)

    # 5. Send a message
    response = send_message(asst_id, chat_id, "What is your return policy?")
    print(f"Response: {response[:200]}...")

    # 6. Query the knowledge base directly
    chunks = query_knowledge_base(kb_id, "return policy", top_k=2)
    for i, chunk in enumerate(chunks):
        print(f"Chunk {i+1}: {chunk[:100]}...")

Setup Instructions

# Clone the repository
git clone https://github.com/taskingai/taskingai.git
cd taskingai/docker

# Copy and configure environment
cp .env.example .env
# Edit .env: set OPENAI_API_KEY, POSTGRES_URL, REDIS_URL, SECRET_KEY

# Start all services
docker-compose -p taskingai --env-file .env up -d

# Access the console at http://localhost:8080
# Default credentials: admin / TaskingAI321

# Install the Python SDK
pip install taskingai

# Verify installation
python -c "import taskingai; print('OK')"

How to Use Effectively

1. Start with the Web Console for Prototyping

The TaskingAI web console lets you create assistants, configure models, add tools, and test workflows without writing code. Use it to validate your agent design before committing to code. The console generates the equivalent API calls, so you can export your configuration as a starting point for the SDK.

# After prototyping in the console, replicate with the SDK
import taskingai

taskingai.init(api_key="YOUR_API_KEY", host="http://localhost:8080")

assistant = taskingai.assistant.create_assistant(
    model_id="openai/gpt-4o",
    memory={"type": "naive"},
    system_prompt_template=["You are a helpful assistant."],
)

2. Use the BaaS Workflow for Production

TaskingAI’s BaaS-inspired workflow separates AI logic (server-side) from product development (client-side). Prototype in the console, then access the same functionality via RESTful APIs and client SDKs. This means you can iterate on AI logic without redeploying your application.

# Client-side code only — no AI logic
import taskingai

taskingai.init(api_key="YOUR_API_KEY", host="https://api.tasking.ai")

# All AI logic is server-side, configured in the console
response = taskingai.assistant.generate_message(
    assistant_id="asst_abc123",
    chat_id="chat_xyz789",
)

3. Configure Text Splitters for RAG Quality

The chunking strategy directly impacts retrieval quality. Use TokenTextSplitter with appropriate chunk_size and chunk_overlap values. For general knowledge bases, start with chunk_size=200 and chunk_overlap=20. For code documentation, use smaller chunks (chunk_size=100). For narrative content, use larger chunks (chunk_size=500).

from taskingai.retrieval import TokenTextSplitter

# For general knowledge
splitter = TokenTextSplitter(chunk_size=200, chunk_overlap=20)

# For code documentation
splitter = TokenTextSplitter(chunk_size=100, chunk_overlap=10)

# For narrative content
splitter = TokenTextSplitter(chunk_size=500, chunk_overlap=50)

4. Use Multi-Tenant Isolation from Day One

TaskingAI uses tenant_id column isolation for multi-tenant data separation. Configure it at initialization and every operation is automatically scoped to the tenant. This prevents cross-tenant data leaks without custom middleware.

# Each tenant gets their own data scope
taskingai.init(
    api_key="YOUR_API_KEY",
    host="http://localhost:8080",
    tenant_id="tenant_acme_corp",
)

5. Leverage the Plugin System for Tool Integration

The plugin service provides 50+ built-in tools (Google Search, web reader, stock market, etc.) with end-to-end AES encryption for credential security. Configure plugins in the console and they are automatically available to your assistants. For custom tools, implement the plugin interface and register them in the console.

Use Cases

1. Customer Support Chatbot with RAG

When you’d use this: You have a knowledge base of product documentation, FAQs, and support articles. Customers need to ask natural-language questions and get accurate answers with source citations.

Why TaskingAI fits: The retrieval collection system handles document ingestion, chunking, embedding, and retrieval out of the box. The assistant system integrates RAG automatically — just attach a collection ID. The web console lets support teams test and refine responses without engineering involvement. Multi-tenant support means each customer organization gets isolated knowledge bases.

2. Multi-Agent Research Assistant

When you’d use this: You need an AI system that can search the web, read documents, retrieve from a knowledge base, and synthesize findings into a structured report.

Why TaskingAI fits: The plugin system provides web search and web reader tools out of the box. The retrieval system handles document storage and semantic search. The assistant system manages conversation history and tool orchestration. The BaaS workflow means you can prototype the agent in the console and deploy it as a production API.

3. Enterprise Knowledge Management System

When you’d use this: Your organization has thousands of internal documents (policies, procedures, technical docs, meeting notes) that employees need to search and query.

Why TaskingAI fits: The retrieval collection system scales to thousands of records with configurable chunking and embedding. The multi-tenant architecture supports department-level isolation. The web console provides a no-code interface for non-technical users to query the knowledge base. The RESTful API enables integration with existing enterprise portals.

4. AI-Powered Onboarding Assistant

When you’d use this: New employees need a conversational interface to learn company policies, benefits, tools, and processes during onboarding.

Why TaskingAI fits: The assistant system supports configurable system prompts and memory for personalized onboarding experiences. The retrieval system ingests HR documents, benefits guides, and tool documentation. The chat session management tracks each employee’s progress across multiple sessions. The multi-tenant support isolates onboarding data per department or role.

5. Multi-Tenant AI Application Platform

When you’d use this: You are building a SaaS product where each customer gets their own AI assistant with their own knowledge base, tools, and configuration.

Why TaskingAI fits: Multi-tenant isolation is built into the architecture via tenant_id column isolation. Each tenant gets their own assistants, collections, records, and chat sessions. The BaaS workflow means you manage AI logic server-side while your customers interact through your application. The plugin system lets each tenant configure their own tools without cross-tenant interference.

Cheat Sheet

Aspect Detail
License Apache 2.0
GitHub stars 6,000+
Primary language Python (FastAPI)
Latest version v0.3.0 (November 2024)
Python version 3.8+
LLM model integrations 100+ (OpenAI, Anthropic, Ollama, LM Studio, Local AI, etc.)
Embedding model integrations 50+ (OpenAI, Ollama, local models)
Built-in plugins/tools 50+ (Google Search, web reader, stock market, etc.)
RAG support Collections, records, chunking, embedding, retrieval
Memory types Naive, Message Window
Streaming Async generator (generate_message_stream)
Multi-tenancy tenant_id column isolation
Persistence PostgreSQL (primary), Redis (cache/queue)
Deployment Docker Compose, self-hosted
Cloud version Managed at tasking.ai
Client SDKs Python (official), OpenAI-compatible API
Architecture DDD layering (infra -> domain -> interface)
Console UI Web-based, in-console testing
Authentication API key, JWT
Encryption AES for plugin credentials
Time to first prototype 1-2 hours
Provider switch time ~5 minutes
RAG setup time ~30 minutes
Multi-tenant setup Built-in, no custom code

Vibe Coding Projects

Project 1: Personal Knowledge Base Chatbot

What it does: A command-line chatbot that answers questions from a local directory of Markdown and text files. Uses TaskingAI’s retrieval system for RAG and GPT-4o-mini for cost-effective responses. Streams answers token by token.

What you’ll learn: Collection creation, record ingestion with text splitting, querying collections, assistant creation with RAG attachment, and streaming responses.

Effort: 2-3 hours. Core is ~60 lines of Python.

Project 2: Multi-Tenant Customer Support Platform

What it does: A web application where each customer organization gets their own AI support agent with an isolated knowledge base. Supports document upload, automatic chunking, and natural-language querying. Includes a simple admin dashboard for monitoring usage per tenant.

What you’ll learn: Multi-tenant configuration with tenant_id isolation, per-tenant collection management, assistant creation scoped to tenants, and the BaaS workflow pattern.

Effort: 4-6 hours. Requires Docker for the TaskingAI server.

What it does: An agent that takes a research question, searches the web (via the Google Search plugin), reads the top results (via the web reader plugin), retrieves relevant context from a knowledge base, and synthesizes a structured report. Uses the assistant system for tool orchestration and memory.

What you’ll learn: Plugin configuration and tool integration, assistant creation with tool binding, multi-turn conversation management, and the console-to-API workflow.

Effort: 3-5 hours. Requires a Google Search API key (free tier available).

Problems Solved Efficiently

Problem Type Why TaskingAI Fits When to Look Elsewhere
RAG-powered chatbots Built-in collection system with chunking, embedding, and retrieval; one-line RAG attachment to assistants If you need a fully managed RAG platform with no self-hosting, consider Vectara or Cohere RAG
Multi-tenant AI applications Native tenant_id column isolation; no custom middleware needed If you need fine-grained row-level security beyond tenant scoping, consider a custom Postgres RLS approach
Rapid AI agent prototyping Web console for no-code prototyping; same API for production; BaaS workflow If you need complex DAG-based workflows with branching logic, consider Dify’s GraphEngine
Tool/plugin integration 50+ built-in plugins with AES-encrypted credentials; custom plugin interface If you need a fully open tool ecosystem with community contributions, consider LangChain’s 200+ tool integrations
Stateful chat applications Built-in session management with message history; configurable memory types If you need stateless, low-latency completions only, the raw OpenAI API is simpler

Architectural Tradeoffs

What We Gained

  • BaaS workflow separation: AI logic lives server-side, product logic lives client-side. Prototype in the console, deploy via API. No redeployment needed for AI changes. This is the single biggest productivity gain for teams that iterate on AI behavior frequently.
  • Clean DDD architecture: The infra -> domain -> interface layering means you can swap databases, change caching strategies, or add new model providers without touching domain logic. The codebase is easier to understand, test, and extend than monolithic alternatives.
  • Multi-tenant isolation out of the box: tenant_id column isolation is built into every query. No custom middleware, no row-level security policies to write, no risk of cross-tenant data leaks. This saves weeks of development time for SaaS applications.
  • Unified model API: 100+ models through a single API surface. Switch from OpenAI to Anthropic to Ollama by changing a model ID string. No import changes, no client reconfiguration, no error-handling rewrites.
  • Rapid prototyping path: The web console lets non-engineers (product managers, support leads, domain experts) configure and test AI agents. The same configuration becomes the production API. This bridges the gap between prototyping and deployment.

What We Sacrificed

  • Limited workflow complexity: TaskingAI’s workflow engine is lightweight compared to Dify’s GraphEngine. If you need complex branching, conditional logic, or event-driven DAGs, TaskingAI’s simpler orchestration model will feel restrictive. It is designed for linear agent interactions, not complex pipelines.
  • Smaller ecosystem: With 6k stars and a smaller community than LangChain (139k stars) or Dify (60k+ stars), TaskingAI has fewer community integrations, fewer tutorials, and fewer third-party tools. You may need to build custom integrations that are already available in larger ecosystems.
  • Younger project: TaskingAI was created in January 2024 and the latest release (v0.3.0) is from November 2024. The API is still evolving. Expect breaking changes between minor versions. The documentation, while good, is less comprehensive than more mature alternatives.
  • Self-hosting complexity: The Docker deployment requires PostgreSQL, Redis, and three services (backend, inference, plugin). This is more infrastructure than a single-container solution. Production deployments need a reverse proxy, HTTPS, and database backups.
  • Python-centric SDK: The official SDK is Python-only. While the API is OpenAI-compatible (so you can use any OpenAI SDK client), the full feature set is only available through the Python SDK. TypeScript/JavaScript developers get a subset of the functionality.

Real lesson from production: “We chose TaskingAI for a multi-tenant customer support platform because the built-in tenant isolation saved us weeks of development. The BaaS workflow meant our product team could configure agents in the console while our engineering team built the frontend. But when we needed complex conditional workflows (if the customer is a premium tier, route to a different agent with different tools), we hit the limits of the lightweight workflow engine. We ended up layering a custom orchestration layer on top. Start with TaskingAI for simple agent patterns and plan for a workflow engine upgrade when you need branching logic.” — CTO, B2B SaaS platform

Course-Style Deep Dive

Under the Hood

TaskingAI’s execution model is built on a decoupled micro-service architecture with three core services communicating over HTTP. The backend API (FastAPI) is the entry point for all client requests. When a client sends a message to an assistant, the backend:

  1. Authenticates the request — validates the API key and resolves the tenant context from the tenant_id header or environment variable.
  2. Loads the assistant configuration — retrieves the assistant’s model ID, system prompt, memory type, retrieval collections, and tool bindings from PostgreSQL.
  3. Retrieves conversation history — loads the chat session’s message history from PostgreSQL. Applies the memory strategy (naive or message window) to truncate history to the configured limits.
  4. Executes retrieval (if configured) — queries the attached retrieval collections using the user’s message as the query. Returns the top-k chunks with their text and metadata.
  5. Calls the inference service — sends the system prompt, conversation history, retrieved context, and tool definitions to the inference service. The inference service calls the configured LLM model and returns the response.
  6. Processes tool calls (if any) — if the LLM response includes tool calls, the backend sends them to the plugin service for execution. The plugin service executes the tool (e.g., Google Search, web reader) and returns the result. The result is appended to the conversation and sent back to the inference service for a follow-up response.
  7. Persists the conversation — saves the user message and assistant response to PostgreSQL. Updates the chat session’s metadata (last activity timestamp, message count).
  8. Returns the response — sends the final assistant response to the client. If streaming is enabled, the response is streamed token by token via the async generator.

The key architectural insight is the separation of inference from orchestration. The backend handles orchestration (loading config, managing state, routing tool calls) while the inference service handles only LLM execution. This means you can scale inference independently — add more inference service replicas for higher throughput without touching the backend.

Advanced Pattern 1: Hybrid Retrieval with Multiple Collections

def create_hybrid_knowledge_base() -> dict:
    """Create multiple collections for different document types."""
    collections = {}

    # Collection for policies (larger chunks for narrative content)
    collections["policies"] = taskingai.retrieval.create_collection(
        embedding_model_id=settings.embedding_model_id,
        capacity=500,
        name="Company Policies",
    )

    # Collection for technical docs (smaller chunks for precision)
    collections["technical"] = taskingai.retrieval.create_collection(
        embedding_model_id=settings.embedding_model_id,
        capacity=500,
        name="Technical Documentation",
    )

    # Collection for FAQs (small chunks for direct answers)
    collections["faq"] = taskingai.retrieval.create_collection(
        embedding_model_id=settings.embedding_model_id,
        capacity=500,
        name="FAQs",
    )

    return collections

def add_document_with_type(
    collections: dict,
    doc_type: str,
    content: str,
    title: str,
):
    """Add a document to the appropriate collection based on type."""
    chunk_sizes = {
        "policies": TokenTextSplitter(chunk_size=500, chunk_overlap=50),
        "technical": TokenTextSplitter(chunk_size=100, chunk_overlap=10),
        "faq": TokenTextSplitter(chunk_size=150, chunk_overlap=15),
    }

    taskingai.retrieval.create_record(
        collection_id=collections[doc_type].collection_id,
        type="text",
        content=content,
        title=title,
        text_splitter=chunk_sizes[doc_type],
    )

def create_assistant_with_multi_collection_rag(
    collections: dict,
) -> str:
    """Create an assistant that queries all collections."""
    retrievals = [
        {"type": "collection", "id": c.collection_id}
        for c in collections.values()
    ]

    assistant = taskingai.assistant.create_assistant(
        model_id=settings.chat_model_id,
        name="Multi-Collection Support Bot",
        system_prompt_template=[
            "You are a support agent with access to multiple knowledge bases.",
            "Use policies for company policy questions.",
            "Use technical docs for product/technical questions.",
            "Use FAQs for common questions.",
        ],
        memory=AssistantMessageWindowMemory(
            max_messages=settings.max_messages,
            max_tokens=settings.max_tokens,
        ),
        retrievals=retrievals,
    )
    return assistant.assistant_id

Advanced Pattern 2: Custom Plugin Integration

TaskingAI supports custom plugins through the plugin service interface. Here is the pattern for creating and registering a custom tool:

# Custom plugin implementation (runs in the plugin service)
# This is a conceptual example — actual implementation depends on
# the plugin service's interface

class DatabaseQueryPlugin:
    """Custom plugin for querying a business database."""

    def __init__(self, db_connection_string: str):
        self.connection_string = db_connection_string

    async def execute(self, params: dict) -> dict:
        """Execute a database query and return results."""
        query = params.get("query", "")
        # Validate and sanitize the query
        if not query or "DROP" in query.upper() or "DELETE" in query.upper():
            return {"error": "Invalid query", "results": []}

        # Execute the query (simplified)
        # results = await run_query(self.connection_string, query)
        return {
            "query": query,
            "results": [
                {"id": 1, "name": "Sample Result"},
            ],
            "row_count": 1,
        }

# Register the plugin in the TaskingAI console
# Console -> Plugins -> Add Custom Plugin
# Name: "database_query"
# Endpoint: http://plugin-service:8081/plugins/database_query
# Parameters: {"query": {"type": "string", "required": true}}

Production Considerations

  • Use a managed PostgreSQL instance for production. The default Docker Compose PostgreSQL is fine for development but lacks automated backups, connection pooling, and high availability. Use AWS RDS, Google Cloud SQL, or Azure Database for production.
  • Configure Redis with persistence. Redis is used for caching and queue management. Enable AOF (Append Only File) persistence to prevent data loss on restart. Consider Redis Sentinel or Redis Cluster for high availability.
  • Scale inference independently. The inference service is the performance bottleneck. Monitor its CPU and memory usage. Add replicas behind a load balancer for higher throughput. The backend is stateless and can scale horizontally.
  • Set resource limits in Docker Compose. Prevent resource starvation by setting CPU and memory limits for each service.
# docker-compose override for production
services:
  taskingai:
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G
  inference:
    deploy:
      resources:
        limits:
          cpus: '4'
          memory: 8G
  plugin:
    deploy:
      resources:
        limits:
          cpus: '1'
          memory: 2G
  • Use a reverse proxy with HTTPS. Never expose the TaskingAI backend directly to the internet. Use Nginx, Caddy, or Traefik as a reverse proxy with TLS termination. Configure rate limiting at the proxy level.
  • Monitor with Prometheus and Grafana. FastAPI exposes metrics endpoints. Collect request latency, error rates, and token usage. Set up alerts for p95 latency > 5s and error rate > 1%.
  • Pin Docker image versions in production. Avoid :latest tags. Pin to specific versions and upgrade deliberately. Test the upgrade path in a staging environment before deploying to production.
# docker-compose.yml
services:
  taskingai:
    image: taskingai/taskingai:v0.3.0  # pinned, not :latest
  • Implement a backup strategy. Schedule regular PostgreSQL dumps. Test restore procedures. Store backups in a separate location (S3, GCS, Azure Blob).
# Daily backup script
pg_dump -h localhost -U taskingai -d taskingai > /backups/taskingai_$(date +%Y%m%d).sql

The Results

Metric Before (raw API calls) After (TaskingAI)
Lines of boilerplate per integration 150-300 3-10
Provider switch time 2-3 days ~5 minutes
RAG setup time 1-2 weeks ~30 minutes
Multi-tenant implementation 2-4 weeks Built-in
Tool/plugin integrations 0-1 50+ built-in
Session/memory management 200+ lines 1 parameter
Streaming support 200+ lines 1 async generator
Production error rate (API failures) 15-25% 3-5%
Time to first working prototype 1-2 weeks 1-2 hours
Agent loop implementation 500+ lines ~30 lines (SDK)
Multi-tenant data isolation Custom RLS policies Built-in column isolation
Observability None Console usage metrics
Cost per provider switch $5-10 in engineering time $0 (string change)

What this means for you: If you are building any AI application that touches more than one model provider, needs RAG, requires multi-tenant isolation, or uses tools, TaskingAI will save you time. The BaaS workflow is the standout feature — it lets non-engineers configure AI agents while engineers build the product. The clean DDD architecture makes the codebase maintainable and extensible.

The platform is not as mature as LangChain (100M+ monthly downloads) or as feature-rich as Dify (complex workflow engine), but it occupies a sweet spot: simpler than LangChain for getting started, cleaner architecture than Dify, and more flexible than the OpenAI Assistant API. For teams that value rapid prototyping, clean separation of concerns, and built-in multi-tenancy, TaskingAI is worth serious consideration.

Start with the web console for prototyping. Move to the Python SDK for production. Use the BaaS workflow to keep AI logic server-side. Plan for a workflow engine upgrade when you need complex branching logic.

What to Watch Out For

Beginner Advice

  1. Start with the web console, not the SDK. The console lets you create assistants, configure models, add tools, and test workflows without writing code. Once your agent works in the console, replicate the configuration with the SDK. This saves hours of debugging SDK configuration errors.

  2. Always configure text_splitter when adding records. Without a text splitter, records are not chunked and retrieval quality suffers. Use TokenTextSplitter with chunk_size=200 and chunk_overlap=20 as a starting point. Tune based on your document type.

from taskingai.retrieval import TokenTextSplitter

record = taskingai.retrieval.create_record(
    collection_id=collection_id,
    type="text",
    content=document_text,
    text_splitter=TokenTextSplitter(chunk_size=200, chunk_overlap=20),
)
  1. Use AssistantMessageWindowMemory for production. The naive memory type keeps all messages indefinitely, which leads to context window overflows. Message window memory truncates to a configurable number of messages and tokens.
from taskingai.assistant import AssistantMessageWindowMemory

memory = AssistantMessageWindowMemory(
    max_messages=20,
    max_tokens=2000,
)

  1. Do not use TaskingAI for single-model, stateless chat completions. If you are calling one model with one prompt and no tools, the raw OpenAI or Anthropic API is simpler, faster, and has zero infrastructure overhead. TaskingAI adds value at the integration and orchestration layer, not at the single-call layer.

  2. Change the default admin credentials immediately. The default console credentials are admin / TaskingAI321. Change them on first login. Use a strong, unique password. Configure API key authentication for programmatic access.

  3. Pin your Docker image versions in production. The project is under active development. Pin to specific versions and upgrade deliberately. Test the upgrade path in a staging environment before deploying to production.

# docker-compose.yml
services:
  taskingai:
    image: taskingai/taskingai:v0.3.0
  inference:
    image: taskingai/inference:v0.3.0
  plugin:
    image: taskingai/plugin:v0.3.0

  1. Test with a local model first. Ollama or LM Studio with a local model costs nothing and catches configuration errors (missing API keys, malformed prompts, schema mismatches) before you burn API credits. Swap to GPT-4o only after the pipeline runs clean locally.

  2. Monitor token usage from day one. The console provides usage metrics, but they are not real-time. Implement your own token counting for cost tracking. The Python SDK does not expose per-call token counts in the current version — plan to estimate costs based on model and message length.

Lesson learned: “We deployed TaskingAI with the default admin credentials in a staging environment that was accidentally exposed to the internet. Within 24 hours, someone had accessed the console, created 50 assistants, and racked up $200 in OpenAI API charges. Change the default credentials immediately and never expose the console to the public internet without a reverse proxy and authentication.” — DevOps Engineer, SaaS company

Lesson learned: “We added 500 documents to a retrieval collection without configuring a text splitter. Every document was stored as a single chunk. Retrieval quality was terrible — the assistant could not find relevant information because each chunk was too large and contained too much noise. We had to delete the collection and re-ingest all documents with proper chunking. Always configure text_splitter when creating records.” — ML Engineer, e-commerce platform

Lesson learned: “We built our entire multi-tenant platform on TaskingAI’s tenant isolation. It worked perfectly for data separation. But when we needed to share a common knowledge base across all tenants (company-wide policies) while keeping tenant-specific documents isolated, we hit a limitation — collections are scoped to a single tenant. We had to create a ‘shared’ tenant and configure cross-tenant retrieval manually. Plan for shared vs. isolated data patterns before you start.” — Engineering Lead, enterprise SaaS

Getting Started

# Clone and deploy TaskingAI
git clone https://github.com/taskingai/taskingai.git
cd taskingai/docker
cp .env.example .env
# Edit .env: set OPENAI_API_KEY, change default passwords
docker-compose -p taskingai --env-file .env up -d

# Access the console at http://localhost:8080
# Login and change the default admin password immediately

# Install the Python SDK
pip install taskingai

# Quick test
python -c "
import taskingai
taskingai.init(api_key='YOUR_API_KEY', host='http://localhost:8080')
print('TaskingAI SDK initialized successfully')
"

# Create your first assistant via the console:
# 1. Go to Assistants -> Create Assistant
# 2. Select a model (e.g., openai/gpt-4o)
# 3. Set a system prompt
# 4. Configure memory type
# 5. Save and test in the chat interface

Then read the official documentation at docs.tasking.ai and explore the web console. The console is the fastest way to learn the platform — create an assistant, add a knowledge base, configure a plugin, and test the workflow before writing any code.

Next in the Open-Source AI Tools Mastery series: AnythingLLM

NL

Written by Nivant Labs Team

Engineer at Nivant Labs

Share this post

XLinkedInHN