Workflow Orchestration Skill

You are an expert in AI workflow orchestration, building the multi-step pipelines

used by Perplexity AI, Cohere, and enterprise research automation systems.

You write production-ready orchestration code in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

Part 1 — LangGraph: Stateful Agent Graphs

# pip install langgraph langchain-anthropic

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from typing import TypedDict, Annotated
import operator


# ── State definition ──────────────────────────────────────────────────────────
class IMIResearchState(TypedDict):
    query: str
    search_results: list[dict]
    analysis: str
    report: str
    messages: Annotated[list, operator.add]  # append-only message list


# ── Node functions ────────────────────────────────────────────────────────────
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")


def search_node(state: IMIResearchState) -> dict:
    """Search for relevant information."""
    from tavily import TavilyClient
    import os
    tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
    results = tavily.search(state["query"], max_results=5)
    return {"search_results": results.get("results", [])}


def analyse_node(state: IMIResearchState) -> dict:
    """Analyse search results for fan intelligence insights."""
    context = "\n".join([r.get("content", "")[:300] for r in state["search_results"][:5]])
    response = llm.invoke([
        {"role": "system", "content": "You are an IMI sports fan intelligence analyst. Use British English."},
        {"role": "user", "content": f"Context:\n{context}\n\nAnalyse for fan intelligence insights on: {state['query']}"}
    ])
    return {"analysis": response.content}


def report_node(state: IMIResearchState) -> dict:
    """Generate final IMI-style research report."""
    response = llm.invoke([
        {"role": "system", "content": "You write IMI research reports in British English."},
        {"role": "user", "content": f"Write a concise research summary based on:\n{state['analysis']}"}
    ])
    return {"report": response.content}


def should_continue(state: IMIResearchState) -> str:
    """Conditional edge: decide next node."""
    if not state.get("search_results"):
        return "search"
    if not state.get("analysis"):
        return "analyse"
    return END


# ── Build graph ───────────────────────────────────────────────────────────────
def build_imi_research_graph() -> StateGraph:
    graph = StateGraph(IMIResearchState)

    graph.add_node("search", search_node)
    graph.add_node("analyse", analyse_node)
    graph.add_node("report", report_node)

    graph.set_entry_point("search")
    graph.add_edge("search", "analyse")
    graph.add_edge("analyse", "report")
    graph.add_edge("report", END)

    return graph.compile()


def run_research_pipeline(query: str) -> dict:
    """Run the full IMI research pipeline."""
    app = build_imi_research_graph()
    result = app.invoke({"query": query, "messages": [], "search_results": [], "analysis": "", "report": ""})
    return {"query": query, "report": result["report"], "analysis": result["analysis"]}

Part 2 — Simple DAG Orchestrator (No Dependencies)

import asyncio, time
from dataclasses import dataclass, field
from typing import Callable, Any
from collections import defaultdict


@dataclass
class Task:
    name: str
    fn: Callable
    dependencies: list[str] = field(default_factory=list)
    result: Any = None
    status: str = "pending"   # pending | running | done | failed
    error: str | None = None


class DAGOrchestrator:
    """
    Lightweight DAG task orchestrator — no Airflow/Prefect required.
    Runs tasks in dependency order, parallelising where possible.
    """

    def __init__(self):
        self.tasks: dict[str, Task] = {}

    def add_task(self, name: str, fn: Callable, dependencies: list[str] = None):
        self.tasks[name] = Task(name=name, fn=fn, dependencies=dependencies or [])

    def _get_ready_tasks(self) -> list[str]:
        """Get tasks whose dependencies are all done."""
        ready = []
        for name, task in self.tasks.items():
            if task.status == "pending":
                deps_done = all(
                    self.tasks[d].status == "done"
                    for d in task.dependencies
                    if d in self.tasks
                )
                if deps_done:
                    ready.append(name)
        return ready

    async def run(self) -> dict[str, Any]:
        """Run all tasks, respecting dependencies."""
        results = {}

        while any(t.status in ("pending", "running") for t in self.tasks.values()):
            ready = self._get_ready_tasks()
            if not ready:
                await asyncio.sleep(0.1)
                continue

            # Run ready tasks concurrently
            async def run_task(name: str):
                task = self.tasks[name]
                task.status = "running"
                try:
                    dep_results = {d: self.tasks[d].result for d in task.dependencies}
                    if asyncio.iscoroutinefunction(task.fn):
                        task.result = await task.fn(**dep_results)
                    else:
                        task.result = task.fn(**dep_results)
                    task.status = "done"
                    results[name] = task.result
                except Exception as e:
                    task.status = "failed"
                    task.error = str(e)

            await asyncio.gather(*[run_task(name) for name in ready])

        return results


# ── IMI research pipeline using DAG ──────────────────────────────────────────
async def run_imi_dag_pipeline(topic: str) -> dict:
    """
    Parallel IMI research pipeline:
    - search + segment_analysis run in parallel (no dependencies)
    - synthesis depends on both
    - report depends on synthesis
    """
    dag = DAGOrchestrator()

    dag.add_task("search", lambda: {"results": [f"Stub result for {topic}"]})
    dag.add_task("segment_analysis", lambda: {"segments": ["Tribal", "Passionate"]})
    dag.add_task("synthesis", lambda search, segment_analysis: {
        "synthesis": f"Synthesised: {search['results']} + {segment_analysis['segments']}"
    }, dependencies=["search", "segment_analysis"])
    dag.add_task("report", lambda synthesis: {
        "report": f"IMI Report: {synthesis['synthesis']}"
    }, dependencies=["synthesis"])

    return await dag.run()

Part 3 — Prefect: Production Workflow Automation

# pip install prefect

from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta


@task(
    retries=3,
    retry_delay_seconds=10,
    cache_key_fn=task_input_hash,
    cache_expiration=timedelta(hours=1)
)
def fetch_research_data(topic: str) -> list[dict]:
    """Fetch web data — auto-retried on failure, cached for 1 hour."""
    logger = get_run_logger()
    logger.info(f"Fetching data for: {topic}")
    # Replace with actual fetch logic
    return [{"title": f"Result for {topic}", "content": "stub content"}]


@task(retries=2)
def analyse_data(data: list[dict], query: str) -> str:
    """Analyse fetched data with Claude."""
    import anthropic
    client = anthropic.Anthropic()
    context = "\n".join([d.get("content", "")[:200] for d in data[:5]])
    resp = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=500,
        messages=[{"role": "user", "content": f"Analyse for IMI research: {query}\n\nData: {context}"}]
    )
    return resp.content[0].text


@task
def generate_report(analysis: str, topic: str) -> str:
    """Generate final report."""
    import anthropic
    client = anthropic.Anthropic()
    resp = client.messages.create(
        model="claude-3-haiku-20240307",
        max_tokens=800,
        messages=[{"role": "user", "content": f"Write a concise IMI research report on '{topic}':\n{analysis}"}]
    )
    return resp.content[0].text


@flow(name="IMI-Research-Pipeline", log_prints=True)
def imi_research_flow(topic: str) -> dict:
    """
    Full IMI research flow — observable, retryable, cacheable.
    Run with: imi_research_flow("football fan sponsorship UK")
    Schedule with Prefect Cloud or self-hosted server.
    """
    data = fetch_research_data(topic)
    analysis = analyse_data(data, topic)
    report = generate_report(analysis, topic)
    return {"topic": topic, "report": report}

Part 4 — Celery: Background Task Queue

# pip install celery redis

from celery import Celery
import os

# ── App setup ─────────────────────────────────────────────────────────────────
app = Celery(
    "imi_tasks",
    broker=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    backend=os.getenv("REDIS_URL", "redis://localhost:6379/0")
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Europe/London",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=300,        # 5 min max
    task_soft_time_limit=240,   # warn at 4 min
)


@app.task(bind=True, max_retries=3, default_retry_delay=30)
def run_research_report(self, topic: str, client_name: str) -> dict:
    """Background task: generate IMI research report."""
    try:
        import anthropic
        client = anthropic.Anthropic()
        resp = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1500,
            messages=[{
                "role": "user",
                "content": f"Write an IMI research report on '{topic}' for {client_name}. British English."
            }]
        )
        return {"status": "complete", "report": resp.content[0].text, "topic": topic}
    except Exception as exc:
        raise self.retry(exc=exc)


def submit_research_job(topic: str, client_name: str) -> str:
    """Submit background research job, return task ID."""
    result = run_research_report.delay(topic, client_name)
    return result.id


def check_job_status(task_id: str) -> dict:
    """Check status of a background job."""
    result = app.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None
    }

Output Standards

• LangGraph: use for complex multi-step agents with loops and conditional branching
• DAG: use for parallel research pipelines without external dependencies
• Prefect: use for scheduled, observable production workflows
• Celery: use for long-running background jobs that shouldn't block the API
• Retries: always set retries=3 with exponential backoff for LLM calls
• Caching: cache task results where inputs are deterministic (Prefect cache_key_fn)
• British English in all log messages, reports, and task names

pip install

pip install langgraph langchain-anthropic prefect celery redis

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)