Metadata-Version: 2.4
Name: ldf-framework
Version: 0.1.0
Summary: LLM Dispatching Framework: async LLM task dispatching and orchestration
Author: Tani
License: MIT
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: openai>=1.0

# LDF — LLM Dispatching Framework

An essential async LLM task dispatching and orchestration library.

This library is designed to be as simple as possible while being used in programming workflows that require both dynamic and deterministic execution paths. 

It provides simple and intuitive usable classes for orchestrating tasks where an LLM is required (e.g. for explainability, data formatting, summarization, routing, etc.). 

The idea behind this library is to use and implement LLM calls only when needed (no agentic loops or tool calling). This framework allows for flexible task orchestration, while keeping token usage predictable and efficient.

It has only one dependency: `openai`.

## Installation

```bash
pip install -e .
```

## Quick Start

```python
import asyncio
from ldf import Task

async def main():
    task = Task("geography", "What is the capital of France?", system_prompt="You are a geography expert.")
    await task.run()
    print(task.output)

asyncio.run(main())
```

### Streaming

```python
async def on_chunk(chunk: str):
    print(chunk, end="", flush=True)

task = Task("story", "Tell me a short story", streaming=True, on_chunk=on_chunk)
await task.run()
```

If `on_chunk` is not provided, this default print behavior is used.

### Iterating on a task (ReAct pattern idea)

`n_iter` feeds each iteration's output back into the LLM as the next prompt:

```python
task = Task("story", "Write one sentence of a story.", n_iter=3)
await task.run()
for step in task.iterations:
    print(step["iteration"], step["output"])
```

`run()` resets `iterations` and `output` at the start of each call, so it's safe to `await` the same `Task`/`Function` instance again later (e.g. reusing module-level task objects across multiple requests) — state from a previous run never leaks into the next one.

### Grouping independent tasks with MultiTask (parallel execution)

`MultiTask` runs its tasks concurrently via `asyncio.gather`. Only use it for tasks that don't depend on each other's output — since they run at the same time, none of them can rely on another's result being in `context` yet:

```python
from ldf import Task, MultiTask

group = MultiTask([
    Task("summary", "Summarize: ..."),
    Task("translation", "Translate to French: ..."),
])

await group.run()
print(group)
```

If you need a strict dependency between tasks (one reads another's output via `context`), don't use `MultiTask` — just list them in order in a `Workflow`'s `tasks`, which always runs sequentially.

### Composing a Workflow

```python
from ldf import Task, MultiTask, Workflow

workflow = Workflow("my_workflow", [
    Task("step_1", "Step 1"),
    MultiTask([Task("step_2a", "Step 2a"), Task("step_2b", "Step 2b")]),
])

await workflow.run()
```

Workflows can also nest other workflows, since a `Workflow` is itself a valid task in a `tasks` list:

```python
sub_workflow = Workflow("sub_workflow", [Task("step_1", "Step 1")])

workflow = Workflow("main_workflow", [
    sub_workflow,
    Task("step_2", "Step 2"),
])

await workflow.run()
```

### Conditions and context

Every `Task`, `MultiTask`, `Workflow`, and `Function` runs against a shared `context` dict, which is passed down through `run()` and populated with each named task's result (`context[task.name] = task`) as it completes.

A `condition` callable receives that context and decides whether the task should run. It can inspect the output of any earlier task by name:

```python
route_task = Task("route", "Classify this request: chat or play_movie")

chat_task = Task(
    "chat", "Reply to the user",
    condition=lambda ctx: ctx["route"].output == "chat",
)
```

If `condition` returns a falsy value, the task is skipped (but still registered in the context under its name). `condition` may also be an async callable.

### Passing one task's result into another

`user_input` accepts a callable (sync or async) instead of a plain string. It's resolved lazily against `context` right before the first iteration runs — so it can reference the output of any task that ran earlier in the same `Workflow`'s `tasks` list:

```python
route_task = Task("route", "Classify this request: chat or play_movie")

extract_task = Task(
    "extract_movie", "Extract the movie title mentioned in this request",
    condition=lambda ctx: ctx["route"].output == "play_movie",
)

summary_task = Task(
    "summary",
    user_input=lambda ctx: f"Write a one-line blurb for the movie: {ctx['extract_movie'].output}",
    condition=lambda ctx: ctx["extract_movie"].output,
)
```

This mirrors `condition`'s pattern exactly (a callable taking `context`, awaited if it returns an awaitable), and only applies to the first iteration's prompt — later `n_iter` iterations still feed off the previous iteration's output. The resolved string is recorded in `task.iterations[0]["input"]` like any other prompt.

### Accessing the original input from anywhere

Pass `user_input` to `Workflow` itself and it's automatically seeded into `context["input"]` before any of its tasks run — so every `condition` or dynamic `user_input` callable in the workflow can read it, with no need to thread it through manually via `run(context=...)`:

```python
workflow = Workflow(
    "chatbot_workflow",
    tasks=[route_task, chat_task, extract_task, play_task],
    user_input=user_input,
)

await workflow.run()
```

```python
chat_task = Task(
    "chat", "...",
    condition=lambda ctx: ctx["route"].output == "chat" and ctx["input"],
)
```

### Running plain functions with Function class

`Function` wraps a regular (sync or async) callable so it can sit inside a `Workflow` alongside `Task`s, with the same `condition`/context support:

```python
from ldf import Function

def play_movie(ctx):
    movie = ctx["extract_movie"].output
    print("Playing " + movie + "...")

play_movie_function = Function(
    name="play_movie",
    func=play_movie,
    condition=lambda ctx: ctx["extract_movie"].output,
)
```

### Tools: LLM-assisted function execution

`Tool` wraps a deterministic function with an LLM-based parameter extraction and result analysis loop. Given a user request, the Tool:

1. Uses an LLM `Task` to extract the function's parameters from the user input
2. Executes the deterministic `function` with those parameters via a `Function`
3. Uses an LLM `Task` to analyze whether the result is valid and useful

If the result is invalid, it retries up to `max_attempts` times.

```python
from ldf import Tool

def extract_ips(text: str) -> str:
    import re
    return str(re.findall(r'\d+\.\d+\.\d+\.\d+', text))

tool = Tool(
    name="ip_extractor",
    description="Extracts all IP addresses from a text. Pass the text to search.",
    function=extract_ips,
    max_attempts=3,
    streaming=True,
)

output = await tool.run("Find IPs in: 192.168.1.1 and 10.0.0.5")
print(output)
```

Optional validators can clean up LLM-extracted parameters or function output:

```python
tool = Tool(
    name="ip_classifier",
    description="Classifies an IP address. Pass a single IP.",
    function=classify_ip,
    parameter_validator=lambda p: p.strip().strip('"'),
    output_validator=lambda o: o.lower(),
)
```

#### Deterministic parameter extraction and result analysis

For most tools, parameters can be extracted deterministically (e.g. with regex) instead of using an LLM call. The `parameter_extractor` and `result_analyzer` callables replace the LLM-based tasks entirely — no tokens consumed, no non-determinism:

```python
import re

tool = Tool(
    name="ip_extractor",
    description="Extracts all IP addresses from a text.",
    function=extract_ips,
    parameter_extractor=lambda user_input: user_input,  # pass input directly
    result_analyzer=lambda output: bool(output and output != "[]"),
)
```

When `parameter_extractor` is provided, the Tool calls it directly instead of running the LLM-based `extract_parameters_task`. When `result_analyzer` is provided, it replaces the LLM-based `analyze_results_task`. Both default to `None`, falling back to LLM — so you can mix deterministic and LLM-based tools in the same agent.

#### Skipping result analysis entirely

For tools where the function output is always valid (e.g. deterministic functions like regex extraction, hashing, arithmetic), set `skip_analysis=True` to return the result immediately after execution — no LLM call, no `result_analyzer`, no retry loop:

```python
tool = Tool(
    name="sha256",
    description="Computes the SHA256 hash of a text.",
    function=compute_hash,
    parameter_extractor=lambda user_input: user_input,
    skip_analysis=True,
)
```

This saves one LLM call per tool execution. The tool still runs parameter extraction and the function itself, then returns the output directly.

### Agents: LLM routing + Tool execution

`Agent` orchestrates multiple `Tool`s with an LLM-based routing loop:

1. An LLM `Task` routes the user input to one or more tools (by name, comma-separated)
2. Each routed tool executes (using its own parameter extraction + execution + analysis loop)
3. An LLM `Task` analyzes whether the combined results fulfill the user's request
4. If not, it re-routes and re-executes up to `max_attempts` times

```python
from ldf import Agent, Tool

agent = Agent(
    name="forensic_agent",
    user_input="Perform a forensic audit on this log: ...",
    tools=[
        Tool(name="ip_extractor", description="Extracts IPs from text", function=extract_ips,
             parameter_extractor=lambda text: text, result_analyzer=lambda o: bool(o)),
        Tool(name="email_extractor", description="Extracts emails from text", function=extract_emails,
             parameter_extractor=lambda text: text, result_analyzer=lambda o: bool(o)),
    ],
    max_attempts=5,
    streaming=True,
)

result = await agent.run()
print(result)
```

Optional `route_validator` and `result_validator` callables can clean up or filter the LLM's routing output and final result:

```python
valid_tools = {"ip_extractor", "email_extractor"}

agent = Agent(
    name="forensic_agent",
    user_input=user_request,
    tools=tools,
    route_validator=lambda route: ", ".join(
        t for t in route.split(",") if t.strip() in valid_tools
    ),
)
```

#### Using MCP servers as tool sources

`Agent` also accepts an `MCP` instance directly as the `tools` argument. The agent automatically calls `parse()` if it hasn't been called yet, and uses the discovered tools:

```python
from ldf import Agent, MCP

mcp = MCP("path/to/mcp.json")
agent = Agent(
    name="my_agent",
    user_input="Read /tmp/example.txt and search for errors",
    tools=mcp,
    max_attempts=5,
    streaming=True,
)

result = await agent.run()
print(result)
```

The agent automatically closes MCP connections when `run()` completes. See the [MCP section](#mcp-model-context-protocol-integration) below for details on configuring MCP servers.

### Retries

Both `Task` and `Function` accept `max_retries` (default `0`, no retry) and `retry_delay` (default `1.0` seconds). On failure they retry with exponential backoff (`retry_delay * 2 ** attempt`) before raising:

```python
task = Task(
    "search", "Look this up...",
    max_retries=3,
    retry_delay=1.0,  # retries after ~1s, 2s, 4s
)
```

Retries apply per LLM/function call, not to the whole `n_iter` loop — if iteration 2 of 3 fails, only that call is retried, not iterations 0-1. Once `max_retries` is exhausted, the original exception propagates as before.

### Logging

`Task`, `Function`, and `LLM` log failures and retry attempts through the standard `logging` module, under logger names `ldf.task`, `ldf.function`, and `ldf.llm`. LDF never configures handlers or exporters itself — that's left entirely to the consuming application, so you can wire it into whatever observability stack you use.

To bridge these logs into OpenTelemetry, attach the OTel SDK's `LoggingHandler` to the root logger (requires `opentelemetry-sdk` and an exporter package — not a dependency of ldf itself):

```python
import logging
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.sdk.resources import Resource

provider = LoggerProvider(resource=Resource.create({"service.name": "my-app"}))
provider.add_log_record_processor(BatchLogRecordProcessor(OTLPLogExporter()))
logging.getLogger().addHandler(LoggingHandler(logger_provider=provider))
```

Or, for local development, just `logging.basicConfig(level=logging.INFO)`.

### Token usage tracking

Token counts are accumulated on each `Task`'s underlying `LLM` client (`llm.input_token`, `llm.output_token`) as calls complete. This works in both non-streaming and streaming modes — in streaming mode, token usage is captured from the final chunk via `stream_options={"include_usage": True}`. `Task`, `MultiTask`, and `Workflow` each expose a `get_token_usage()` method returning:

```python
{
    "input_token_usage": ...,
    "output_token_usage": ...,
    "total_token_usage": ...,
}
```

For `MultiTask` and `Workflow`, this is a recursive sum over `self.tasks` — so a `Workflow` containing nested `Workflow`s or `MultiTask`s reports combined usage for the whole tree:

```python
await workflow.run()
print(workflow.get_token_usage())
# {"input_token_usage": 128, "output_token_usage": 342, "total_token_usage": 470}
```

`Function` doesn't implement `get_token_usage`, since it wraps a plain callable with no `LLM` attached. The aggregation in `MultiTask`/`Workflow` skips any task that lacks the method (via `hasattr`), so mixing in `Function`s or plain callables just contributes zero rather than raising an error.

## Configuration

The `LLM` client is configured via environment variables or defined constructor arguments, so you can personalize it for each task:

| Variable | Default |
|---|---|
| `LLM_MODEL` | `gemma-26b-q5` |
| `LLM_BASE_URL` | `http://localhost:8080/v1` |
| `LLM_API_KEY` | *(empty)* |
| `LLM_SYSTEM_PROMPT` | `You are a helpful assistant.` |
| `LLM_TEMPERATURE` | `0.7` |
| `LLM_MAX_TOKENS` | `1000` |
| `LLM_TIMEOUT` | `60` (seconds) |

Constructor arguments take precedence over environment variables:

```python
from ldf import LLM, Task

llm = LLM(model="gpt-4o-mini", temperature=0.2)
task = Task("greeting", "Hello!", llm=llm)
```

### Sharing an HTTP connection pool across tasks

By default, each `LLM` instance builds its own `openai.AsyncOpenAI` client — fine for a script, but wasteful under load (e.g. many requests/tasks per process each opening their own connection pool). Pass a pre-built `client` to reuse one connection pool across many independently-configured `LLM`/`Task` instances:

```python
import openai
from ldf import LLM, Task

shared_client = openai.AsyncOpenAI(api_key="...", base_url="...", timeout=60)

fast_task = Task("route", "...", llm=LLM(client=shared_client, model="gpt-4o-mini", temperature=0.0))
smart_task = Task("answer", "...", llm=LLM(client=shared_client, model="gpt-4o", system_prompt="Be thorough."))
```

Both tasks share the same underlying HTTP connections, but keep fully independent config *and* independent token counters — `get_token_usage()` is tracked per `LLM` instance, not per client, so sharing the connection pool never mixes up one task's usage with another's. If you don't pass `client`, you get today's behavior (a dedicated client per `LLM` instance) — it's opt-in, not a requirement.

> **Tip:** share the `client`, not the `LLM` instance itself. Passing the *same* `LLM` object to multiple `Task`s is unsafe — `Task.__init__` sets `llm.system_prompt` directly, so two tasks sharing one `LLM` with different `system_prompt`s will clobber each other's setting. Give each `Task` its own `LLM(client=shared_client, ...)` wrapper, as shown above, and only the connection pool gets shared.

## MCP (Model Context Protocol) Integration

`MCP` parses MCP server configurations and creates LDF `Tool` instances for each tool discovered on the servers. It uses only stdlib (`json`, `subprocess`, `select`, `urllib`) — no external dependencies.

### Input formats

`MCP` accepts three input formats:

- **File path** to an `mcp.json` config file
- **JSON string** with server configuration
- **URL** to an MCP server (HTTP transport)

### Transports

Two transports are supported:

- **stdio**: spawns a local process, communicates via newline-delimited JSON-RPC over stdin/stdout
- **HTTP**: sends JSON-RPC POST requests to a remote URL, tracks `Mcp-Session-Id` headers

### Config format

The config follows the standard `mcpServers` schema used by Claude Code, Cursor, and other tools:

```json
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
      "env": {}
    },
    "sentry": {
      "url": "https://mcp.sentry.dev/sse",
      "headers": {
        "Authorization": "Bearer ${SENTRY_TOKEN}"
      }
    }
  }
}
```

Also supports the `servers` key (used by VS Code) as an alternative to `mcpServers`.

### Usage

```python
from ldf import MCP, Agent

mcp = MCP("path/to/mcp.json")
tools = mcp.parse()

for tool in tools:
    print(f"  {tool.name}: {tool.description[:80]}")

agent = Agent(
    name="my_agent",
    user_input="Read /tmp/example.txt and search for errors",
    tools=tools,
    max_attempts=5,
    streaming=True,
)

result = await agent.run()
```

Or pass the `MCP` instance directly to `Agent`:

```python
mcp = MCP("path/to/mcp.json")
agent = Agent(name="my_agent", user_input="...", tools=mcp)
result = await agent.run()
```

When using `MCP` with `Agent`, connections are closed automatically after `run()`. When using `MCP` standalone, `close()` is called automatically on garbage collection (`__del__`), but you can still call it explicitly if needed.

`parse()` performs the MCP handshake (`initialize` → `notifications/initialized` → `tools/list`) for each server and creates an LDF `Tool` for each discovered tool. Each tool's `function` callable handles parameter parsing and `tools/call` invocation via the appropriate transport.

`close()` terminates any spawned stdio processes. It's called automatically when used with `Agent` (after `run()`) or on garbage collection (`__del__`). You only need to call it manually if using `MCP` standalone and want deterministic cleanup.

## Architecture

```
ldf/
├── __init__.py
├── agent.py            # Agent: LLM routing + multi-tool orchestration loop
├── tool.py             # Tool: LLM-assisted function execution with parameter extraction + result analysis
├── mcp.py              # MCP: Model Context Protocol parser, creates Tools from MCP servers
└── core/
    ├── llm.py          # LLM: OpenAI-compatible async client with token tracking
    ├── task.py         # Task: single LLM invocation with iterations
    ├── multitask.py    # MultiTask: concurrent (parallel) task groups
    ├── workflow.py     # Workflow: named sequence of tasks/task groups (can nest workflows)
    └── function.py     # Function: wraps a plain callable with condition/context support
```
