Async Patterns with FFAI

In this tutorial you will use FFAI’s async API to run LLM calls and RAG operations concurrently. By the end you will be able to use async clients, async RAG methods, and async DAG execution.

Prerequisites

  • Python >= 3.10

  • pip install "ffai[rag]"

  • A Mistral API key (or any LiteLLM-supported provider) set as MISTRAL_API_KEY

  • Familiarity with the quickstart and Building a RAG Pipeline with FFAI

Step 1: Create an async client

Use AsyncFFLiteLLMClient instead of FFLiteLLMClient:

import os

from ffai.Clients import AsyncFFLiteLLMClient
from ffai.FFAI import FFAI

client = AsyncFFLiteLLMClient(
    model_string="mistral/mistral-small-latest",
    api_key=os.environ["MISTRAL_API_KEY"],
)

ffai = FFAI(client)

Async clients support the same interface as sync clients. generate_response still works synchronously, but execute_graph and async RAG methods require an async client.

Step 2: Async RAG operations

FFAI provides async counterparts for all RAG methods. Use await with aquery, asearch, and aindex:

from ffai.rag import RAG
from ffai.rag.embed import Embeddings
from ffai.rag.store import VectorStore

embed = Embeddings("mistral/mistral-embed", api_key=os.environ["MISTRAL_API_KEY"])
store = VectorStore(collection_name="async_kb", dir="./async_db")
rag = RAG(embed=embed, store=store, chunk_size=500, chunk_overlap=100)

ffai = FFAI(client, rag=rag)

# Async index
count = await ffai.aindex("Python is a high-level language.", source="doc1")
print(f"Indexed {count} chunks")
# Indexed 1 chunks

# Async search
hits = await ffai.asearch("What is Python?", top_k=5)
print(f"Found {len(hits)} hits")
# Found 1 hits

# Async query
result = await ffai.aquery("What is Python?")
print(result.answer)
print(result.sources)

Async methods return the same types as their sync counterparts: aindex → int, asearch → list[SearchHit], aquery → QueryResult.

Step 3: Async DAG execution

execute_graph runs prompts in topological-parallel order using asyncio.gather. It requires an AsyncFFLiteLLMClient:

prompts = [
    {"prompt_name": "topic", "prompt": "Suggest a topic"},
    {"prompt_name": "outline", "prompt": "Create an outline about {{topic.response}}",
     "history": ["topic"]},
    {"prompt_name": "article", "prompt": "Write an article based on:\n{{outline.response}}",
     "history": ["outline"]},
]

graph_result = await ffai.execute_graph(prompts, max_concurrency=10)

for name, r in graph_result.results.items():
    print(f"{name}: {r.status} ({r.duration_ms:.0f}ms)")

Output (illustrative):

topic: success (842ms)
outline: success (1203ms)
article: success (2156ms)

max_concurrency controls how many prompts run in parallel at each level.

Step 4: Parallel execution with fan-out

When multiple prompts at the same level have no dependencies on each other, they execute concurrently:

prompts = [
    {"prompt_name": "topic", "prompt": "Suggest a technology topic"},
    {"prompt_name": "angles", "prompt": "List 3 angles on {{topic.response}}",
     "history": ["topic"]},
    {"prompt_name": "audience", "prompt": "Describe the audience for {{topic.response}}",
     "history": ["topic"]},
    {"prompt_name": "article", "prompt":
        "Write about {{topic.response}}\nAngles: {{angles.response}}\nAudience: {{audience.response}}",
     "history": ["angles", "audience"]},
]

graph_result = await ffai.execute_graph(prompts, max_concurrency=10)

print(f"Success: {graph_result.success_count}")
print(f"Failed: {graph_result.failed_count}")

Here angles and audience run in parallel at level 1, since neither depends on the other.

Complete listing

import os

from ffai.Clients import AsyncFFLiteLLMClient
from ffai.FFAI import FFAI
from ffai.rag import RAG
from ffai.rag.embed import Embeddings
from ffai.rag.store import VectorStore

client = AsyncFFLiteLLMClient(
    model_string="mistral/mistral-small-latest",
    api_key=os.environ["MISTRAL_API_KEY"],
)

# Async RAG
embed = Embeddings("mistral/mistral-embed", api_key=os.environ["MISTRAL_API_KEY"])
store = VectorStore(collection_name="async_kb", dir="./async_db")
rag = RAG(embed=embed, store=store, chunk_size=500, chunk_overlap=100)
ffai = FFAI(client, rag=rag)

count = await ffai.aindex("Python is a versatile language.", source="doc1")
hits = await ffai.asearch("What is Python?", top_k=5)
result = await ffai.aquery("What is Python?")
print(result.answer)

# Async DAG
prompts = [
    {"prompt_name": "topic", "prompt": "Suggest a topic"},
    {"prompt_name": "outline", "prompt": "Outline about {{topic.response}}",
     "history": ["topic"]},
    {"prompt_name": "article", "prompt": "Article from:\n{{outline.response}}",
     "history": ["outline"]},
]
graph_result = await ffai.execute_graph(prompts, max_concurrency=10)
for name, r in graph_result.results.items():
    print(f"{name}: {r.status}")

Next steps