TaskFlow

​

Overview

TaskFlow is a flexible, agent-based orchestration utility designed for chaining and executing multiple AI agents in a structured, sequential, or parallel fashion.

It supports:

• Step-based and parallel execution of agents
• Dynamic input mapping with references to prior outputs
• Retry policies per step or globally
• Output tracking and structured result return
• Optional callbacks on step completion

TaskFlow is a powerful orchestration engine designed to coordinate multiple AI agents and tools in a structured sequence or parallel execution pattern. It enables complex LLM pipelines with dynamic input resolution, retries, and step-by-step control.

​

✨ Features

• Step-by-step agent orchestration
• Input templating using {{agent_name.key}}
• Parallel execution support
• Retry policy for each step
• Optional callbacks after steps
• Full flow trace access

​

🧠 Example Use Case

Let’s define a flow to extract information and summarize it.

​

1. Define Agents

from chainless import Agent, Tool

def fetch_data(topic: str):
    return f"Data about {topic}"

fetch_tool = Tool("Fetcher", "Fetches raw info", fetch_data)

fetch_agent = Agent(
    name="FetcherAgent",
    llm=llm,
    tools=[fetch_tool],
    system_prompt="Use the tool to fetch data about a topic."
)

summary_agent = Agent(
    name="SummaryAgent",
    llm=llm,
    system_prompt="Summarize the input in 3 lines."
)

​

2. Compose the TaskFlow

from chainless import TaskFlow

flow = TaskFlow(name="InfoFlow")

flow.add_agent("fetch", fetch_agent)
flow.add_agent("summary", summary_agent)

flow.step("fetch", {"topic": "{{input}}"})
flow.step("summary", {"input": "{{fetch.output}}"})

​

3. Run It

result = flow.run("quantum computing")
print(result["output"])

​

🧩 Dynamic Input Mapping

You can reference any previous step’s output using:

• {{input}}: Original input
• {{agent_name.key}}: Output from a specific agent

Example:

flow.step("analyze", {"text": "{{fetch.result}}"})

​

🔁 Retry on Failure

Set retry policy for resilience:

flow.step("summary", {"input": "{{fetch.output}}"}, retry_on_fail=2)

Or globally:

flow = TaskFlow(name="RobustFlow", retry_on_fail=3)

​

⚡ Parallel Execution

Run steps concurrently if independent:

flow.parallel(["fetch_news", "fetch_social"])

​

🔔 Callbacks

Hook into each step:

def log_step(name, output):
    print(f"[{name}] -> {output}")

flow = TaskFlow(name="LoggedFlow", on_step_complete=log_step)

​

🧪 Output Structure

{
  "flow": {
    "fetch": { "output": "Data about quantum computing" },
    "summary": { "output": "Quantum computing is a..." }
  },
  "output": "Quantum computing is a..."
}

​

🆕 New Features

​

🔁 Advanced Step Controls

Each step() now supports fine-grained execution control:

• step_name: Define a custom name for the step (useful for reusing agents).
• timeout: Set a timeout in seconds for the step.
• on_start, on_complete, on_error: Step-level hooks for better control and observability.
• condition: A boolean-returning function to decide whether the step should run.
• depends_on: Specify other steps that must complete before this one runs.

Example:

flow.step(
    "summarizer",
    input_map={"text": "{{input}}"},
    step_name="brief_summary",
    timeout=10,
    on_start=lambda: print("Starting summary..."),
    on_complete=lambda result: print(f"Done: {result}"),
    condition=lambda ctx: ctx["input"] is not None,
    depends_on=["fetch"]
)

​

🏷️ Alias Creation

You can now define aliases for specific keys in a step’s output. This makes your input mappings cleaner and more maintainable.

Usage:

flow.alias("summary", from_step="summarizer", key="text")
flow.step("translator", input_map={"text": "{{summary}}"})

Instead of referencing {{summarizer.text}}, you can now use the simplified {{summary}}.

​

✅ Best Practices

• Keep agents focused (1 purpose per agent)
• Reuse tools through agents
• Use retries on weak/volatile steps
• Combine TaskFlow with UI or APIs

​

🛠 Tool-Based Agent Example

from chainless import Tool, Agent

add_tool = Tool("Adder", "Adds two numbers", lambda a, b: a + b)

math_agent = Agent(
    name="MathAgent",
    llm=llm,
    tools=[add_tool],
    system_prompt="Use the Adder tool to calculate the sum."
)

flow = TaskFlow("MathFlow")
flow.add_agent("math", math_agent)
flow.step("math", {"a": 5, "b": 10})

​

📘 Interface Summary (TS)

interface TaskFlow {
  name: string;
  add_agent(name: string, agent: AgentProtocol): void;
  step(agent_name: string, input_map: Record<string, any>, retry_on_fail?: number): void;
  parallel(agent_names: string[]): void;
  run(user_input: string): {
    flow: Record<string, any>;
    output: any;
  };
}

​

🎯 Use Case Ideas

TaskFlow brings composability and robustness to AI-based systems. Use it to turn isolated agents into scalable intelligent workflows.