Streaming & Async Skill

Role

You are an elite async Python engineer. You build high-throughput, non-blocking

AI systems that handle hundreds of concurrent LLM calls without blocking.

You know every asyncio pattern, every streaming protocol, and how to get

10x throughput vs synchronous code.

Part 1: Async LLM Clients

Async Claude

import asyncio
import anthropic
from typing import AsyncGenerator

async_claude = anthropic.AsyncAnthropic()

async def aclaude(prompt: str,
                   system: str = None,
                   model: str = "claude-haiku-4-5-20251001",
                   max_tokens: int = 2048) -> str:
    """Async Claude call."""
    messages = [{"role": "user", "content": prompt}]
    kwargs = {"model": model, "max_tokens": max_tokens, "messages": messages}
    if system:
        kwargs["system"] = system

    response = await async_claude.messages.create(**kwargs)
    return response.content[0].text


async def aclaude_stream(prompt: str,
                          system: str = None,
                          model: str = "claude-haiku-4-5-20251001") -> AsyncGenerator[str, None]:
    """Async streaming Claude — yields tokens as they arrive."""
    messages = [{"role": "user", "content": prompt}]
    kwargs = {"model": model, "max_tokens": 4096, "messages": messages}
    if system:
        kwargs["system"] = system

    async with async_claude.messages.stream(**kwargs) as stream:
        async for text in stream.text_stream:
            yield text

Async OpenAI

from openai import AsyncOpenAI

async_oai = AsyncOpenAI()

async def agpt(prompt: str,
                system: str = "You are a helpful assistant.",
                model: str = "gpt-4o-mini") -> str:
    """Async GPT call."""
    response = await async_oai.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": prompt}
        ],
        temperature=0.0
    )
    return response.choices[0].message.content

Part 2: Concurrent Batch Processing

The key to high throughput — process many items in parallel.

import asyncio
from typing import Callable, TypeVar
from tqdm.asyncio import tqdm as async_tqdm

T = TypeVar("T")

async def process_concurrent(
    items: list[T],
    async_fn: Callable,
    max_concurrent: int = 10,
    show_progress: bool = True
) -> list:
    """
    Process items concurrently with a semaphore rate limit.
    Used by Perplexity to handle thousands of simultaneous queries.
    """
    semaphore = asyncio.Semaphore(max_concurrent)
    results = [None] * len(items)

    async def _process_with_sem(idx: int, item: T):
        async with semaphore:
            try:
                results[idx] = await async_fn(item)
            except Exception as e:
                logger.error(f"Item {idx} failed: {e}")
                results[idx] = None

    tasks = [_process_with_sem(i, item) for i, item in enumerate(items)]

    if show_progress:
        await async_tqdm.gather(*tasks, desc="Processing")
    else:
        await asyncio.gather(*tasks)

    return results


async def batch_embed_async(texts: list[str],
                             batch_size: int = 100,
                             max_concurrent: int = 5) -> list[list[float]]:
    """
    Async parallel embedding — 5x faster than sequential.
    """
    async_oai = AsyncOpenAI()

    async def _embed_batch(batch: list[str]) -> list[list[float]]:
        response = await async_oai.embeddings.create(
            input=batch,
            model="text-embedding-3-small"
        )
        return [r.embedding for r in response.data]

    # Create batches
    batches = [texts[i:i+batch_size] for i in range(0, len(texts), batch_size)]

    # Process batches concurrently
    batch_results = await process_concurrent(
        batches, _embed_batch, max_concurrent=max_concurrent
    )

    # Flatten
    return [emb for batch in batch_results if batch for emb in batch]

Part 3: Streaming Server with FastAPI

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio, json

app = FastAPI()

@app.post("/chat/stream")
async def stream_chat(request: Request):
    """
    SSE streaming endpoint — sends LLM tokens as Server-Sent Events.
    This is how Claude.ai and ChatGPT serve streaming responses.
    """
    body = await request.json()
    prompt = body.get("prompt", "")
    system = body.get("system", None)

    async def generate():
        try:
            async for token in aclaude_stream(prompt, system=system):
                # SSE format: "data: {json}\n\n"
                yield f"data: {json.dumps({'token': token, 'done': False})}\n\n"

            # Final done event
            yield f"data: {json.dumps({'token': '', 'done': True})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'error': str(e), 'done': True})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )


@app.post("/research/batch")
async def batch_research(request: Request):
    """
    Process multiple research queries concurrently.
    Returns all results when the slowest completes.
    """
    body = await request.json()
    queries = body.get("queries", [])

    async def research_one(query: str) -> dict:
        answer = await aclaude(query, system="You are an IMI research analyst.")
        return {"query": query, "answer": answer}

    results = await process_concurrent(queries, research_one, max_concurrent=5)
    return {"results": results}

Part 4: Async Pipeline Patterns

import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

@dataclass
class PipelineItem:
    id: str
    data: str
    result: str = ""


async def async_pipeline(items: list[str],
                          stages: list[Callable]) -> list[dict]:
    """
    Multi-stage async pipeline.
    Each stage processes all items before the next stage begins.
    """
    current = [PipelineItem(id=str(i), data=item) for i, item in enumerate(items)]

    for stage_fn in stages:
        current = await process_concurrent(current, stage_fn, max_concurrent=10)

    return [{"id": item.id, "result": item.result} for item in current]


async def async_producer_consumer(
    source: AsyncIterator,
    processor: Callable,
    sink: Callable,
    queue_size: int = 10
) -> None:
    """
    Producer-consumer pattern for streaming document processing.
    Producer feeds queue; consumer processes concurrently.
    """
    queue = asyncio.Queue(maxsize=queue_size)

    async def producer():
        async for item in source:
            await queue.put(item)
        await queue.put(None)  # Sentinel

    async def consumer():
        while True:
            item = await queue.get()
            if item is None:
                break
            result = await processor(item)
            await sink(result)
            queue.task_done()

    await asyncio.gather(producer(), consumer())

Part 5: Rate Limiter for API Calls

import asyncio, time
from collections import deque

class AsyncRateLimiter:
    """
    Token bucket rate limiter for async LLM API calls.
    Prevents 429 errors while maximising throughput.
    """

    def __init__(self, requests_per_minute: int = 60):
        self.rpm = requests_per_minute
        self.min_interval = 60.0 / requests_per_minute
        self._request_times: deque = deque()
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        """Wait until we're within rate limits."""
        async with self._lock:
            now = time.time()

            # Remove requests older than 1 minute
            while self._request_times and self._request_times[0] < now - 60:
                self._request_times.popleft()

            # Wait if at limit
            if len(self._request_times) >= self.rpm:
                wait_time = 60 - (now - self._request_times[0])
                if wait_time > 0:
                    await asyncio.sleep(wait_time)

            self._request_times.append(time.time())

    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, *args):
        pass


# Usage
rate_limiter = AsyncRateLimiter(requests_per_minute=50)

async def rate_limited_claude(prompt: str) -> str:
    async with rate_limiter:
        return await aclaude(prompt)

Output Standards

• Use asyncio.Semaphore(10-20) for concurrent API calls — never unlimited
• Always implement rate limiting for production API usage
• Use SSE (text/event-stream) for streaming LLM responses to browsers
• Test concurrent throughput: asyncio.gather(*[task() for _ in range(100)])
• Async batch processing is 5-20x faster than sequential for LLM calls
• Never block the event loop: use await asyncio.sleep() not time.sleep()

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
)