LangTask

LLM-powered parallel multi-task execution — bring your own LLM.

LangTask takes a natural language request, uses an LLM to decompose it into a dependency-aware task graph (DAG), executes independent tasks in parallel using a Rust-backed scheduler, then synthesises the results into a final answer. Works with OpenAI, Anthropic, Google, Mistral, Cohere, Ollama, or any LangChain-compatible LLM.

User request ──► Planner (LLM) ──► DAG (rustworkx)
                                         │
                        ┌────────────────┼────────────────┐
                    Wave 1           Wave 2           Wave N
                  [t1][t2][t3]      [t4][t5]          [t6]   ← parallel
                        └────────────────┼────────────────┘
                                         │
                                  Aggregator (LLM) ──► Final answer

Installation

# Core only (runs with MockLLM for demos):
pip install langtask

# With your chosen LLM provider:
pip install langtask[openai]       # OpenAI / Azure
pip install langtask[anthropic]    # Anthropic Claude
pip install langtask[google]       # Google Gemini
pip install langtask[mistral]      # Mistral
pip install langtask[ollama]       # Ollama (local models)

# Everything:
pip install langtask[all]

Quick Start

from langtask import build_and_run

# No API key needed — uses MockLLM for demo:
state = build_and_run(
    "Give me a full analytics report: total users, Q3 revenue, and churn vs benchmarks.",
    show_telemetry=True,   # optional: print timing + token usage
)
print(state.final_response)

With a real LLM

# OpenAI
from langchain_openai import ChatOpenAI
from langtask import build_and_run

state = build_and_run(
    "Research the latest AI papers and compare with competitor products.",
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    show_telemetry=True,
)

# Anthropic
from langchain_anthropic import ChatAnthropic

state = build_and_run(
    "Analyse our Q3 revenue, active users, and churn rate.",
    llm=ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0),
    show_telemetry=True,
)

# Google Gemini
from langchain_google_genai import ChatGoogleGenerativeAI

state = build_and_run(
    "Summarise findings from three research sources.",
    llm=ChatGoogleGenerativeAI(model="gemini-1.5-pro"),
    show_telemetry=True,
)

# Ollama (local)
from langchain_ollama import ChatOllama

state = build_and_run(
    "Fetch metrics and generate a report.",
    llm=ChatOllama(model="llama3"),
)

Custom Tools

from langtask import build_and_run, register_tool, ToolResult
import requests

def my_crm_tool(inputs: dict) -> ToolResult:
    """Fetch contacts from a CRM API."""
    org_id = inputs.get("org_id", "default")
    data   = requests.get(f"https://api.mycrm.com/contacts?org={org_id}").json()
    return ToolResult(
        output={"contacts": data["total"]},
        tokens_used=0,
    )

register_tool("crm_lookup", my_crm_tool)

state = build_and_run(
    "Look up contacts for org 123 and summarise the data.",
    llm=my_llm,
    show_telemetry=True,
)

The planner LLM will automatically know about crm_lookup because registered tools are passed in the system prompt.

Telemetry Output

When show_telemetry=True, LangTask prints:

────────────────────────────────────────────────────────
  TELEMETRY REPORT
────────────────────────────────────────────────────────
  Total time         :      312 ms
  ├─ Planner         :       48 ms
  ├─ Tool waves      :      238 ms
  │   Wave 1         :      103 ms
  │   Wave 2         :       81 ms
  │   Wave 3         :       54 ms
  └─ Aggregator      :       26 ms
  Total tokens       :      1,490
  ├─ Planner tokens  :        730
  ├─ Tool tokens     :         17
  └─ Aggregator toks :        760
  LLM calls          :          2
  Tool calls         :          7
  Cache hits         :          0
  Execution waves    :          3
────────────────────────────────────────────────────────

Telemetry is also available programmatically:

state = build_and_run("...", show_telemetry=False)
tel   = state.telemetry

print(tel.total_tokens)        # int
print(tel.total_elapsed_ms)    # float (ms)
print(tel.wave_elapsed_ms)     # list[float] — one entry per wave
print(tel.cache_hits)          # int
print(tel.display())           # formatted string

CLI

# Demo run:
langtask "Analyse our Q3 metrics" --telemetry

# With a provider:
langtask "Research AI trends" --provider anthropic --model claude-sonnet-4-20250514 --telemetry

# Quiet mode (only final answer):
langtask "Summarise our data" --provider openai --quiet

API Reference

build_and_run(user_request, llm=None, scenario="analytics", show_telemetry=False, verbose=True, max_workers=None) → GraphState

register_tool(name, fn)

Register a custom tool. fn receives a dict and must return a ToolResult.

GraphState

Publishing to PyPI

pip install build twine
python -m build
twine upload dist/*

License

MIT