PrashFlow

PrashFlow is a general-purpose Python AI application and agent runtime designed to make Chat, RAG, agents, multi-agent systems, tools, MCP, memory, streaming, and multiple model providers available through a small Python API.

It is not only a RAG library.

What you can build

• Local Ollama RAG applications
• Streaming chat with user sessions
• Semantic, BM25/keyword, hybrid, MMR and multi-query retrieval
• Chroma, Qdrant, FAISS, PGVector and in-memory vector stores
• PDF/DOCX/TXT/Markdown/CSV/web ingestion
• SQL database ingestion
• Single tool-using agents with LangGraph
• Agentic Ollama chat
• Supervisor, sequential and parallel multi-agent workflows
• Custom Python tools
• Optional web search
• MCP configuration/adapter boundary
• OpenAI and OpenAI-compatible models
• LiteLLM model gateway
• Environment-variable based YAML configuration
• Retries and clear validation errors

Architecture

                         PRASHFLOW
                             |
       +---------------------+----------------------+
       |                     |                      |
      Chat                   RAG                  Agent
       |                     |                      |
   Streaming           Ingestion/Retrieval      Tools
   Sessions                  |                    MCP
       |              +-------+-------+             |
       |              |       |       |             |
       |           Semantic  BM25   MMR             |
       |              |       |       |             |
       |              +-------+-------+             |
       |                      |                     |
       |                     RRF                    |
       |                      |                     |
       |                   Rerank                   |
       |                      |                     |
       +----------------------+---------------------+
                              |
                         Model Layer
                              |
                    +---------+---------+
                    |         |         |
                  Ollama    OpenAI    LiteLLM
                              |
                         MultiAgent
                              |
                 +------------+------------+
                 |            |            |
             Supervisor   Sequential    Parallel

Installation

Basic:

pip install prashflow

All optional integrations:

pip install "prashflow[all]"

For local Ollama + Chroma RAG, the all extra is convenient. You can also install only the extras you need.

1. Local Ollama RAG in a few lines

Put documents in ./knowledge:

my-app/
├── knowledge/
│   ├── deployment.pdf
│   ├── architecture.docx
│   ├── troubleshooting.txt
│   └── security.md
├── data/
└── app.py

Pull local models:

ollama pull qwen3:8b
ollama pull nomic-embed-text

Python:

from prashflow import RAG

rag = RAG(
    llm="ollama:qwen3:8b",
    embeddings="ollama:nomic-embed-text",
    vector_db="chroma:./data/chroma",
    retrieval={
        "type": "hybrid",
        "top_k": 5,
        "candidate_k": 20,
        "semantic_weight": 0.6,
        "keyword_weight": 0.4,
    },
)

rag.ingest("./knowledge")

print(rag.ask("What is our production deployment process?"))

PrashFlow hides the LangChain, Chroma, loader and embedding implementation from the application code.

2. Search algorithms

Semantic:

rag.search("production deployment", search_type="semantic")

BM25/keyword:

rag.search("JIRA-12345", search_type="keyword")

MMR:

rag.search("deployment architecture", search_type="mmr")

Hybrid:

rag.search("production deployment", search_type="hybrid")

Multi-query:

rag.search("How do we release an application?", search_type="multi_query")

Hybrid combines semantic and keyword rankings with reciprocal-rank fusion. MMR adds diversity. An optional cross-encoder reranker can be enabled with the rerank extra.

3. Persistent Chroma

rag = RAG(
    llm="ollama:qwen3:8b",
    embeddings="ollama:nomic-embed-text",
    vector_db={
        "provider": "chroma",
        "path": "./data/chroma",
        "collection": "company_docs",
    },
)

The Chroma data remains on disk after the Python process exits.

In-memory vector store

rag = RAG(
    llm="ollama:qwen3:8b",
    embeddings="ollama:nomic-embed-text",
    vector_db={"provider": "memory"},
)

This is intended for tests, demos and short-lived applications.

4. User session + streaming RAG chat

This is the recommended API for a local RAG chatbot:

from prashflow import RAG

rag = RAG(
    llm="ollama:qwen3:8b",
    embeddings="ollama:nomic-embed-text",
    vector_db="chroma:./data/chroma",
    retrieval={"type": "hybrid", "top_k": 5},
)

rag.ingest("./knowledge")

session_id = "user-001"

while True:
    question = input("You: ")
    if question.lower() in {"exit", "quit"}:
        break

    print("AI: ", end="")
    for token in rag.chat_stream(
        session_id=session_id,
        query=question,
        search_type="hybrid",
    ):
        print(token, end="", flush=True)
    print()

The session stores conversation history independently for each session_id.

user-001 -> conversation A
user-002 -> conversation B
user-003 -> conversation C

The default session backend is in-memory. A persistent Redis/PostgreSQL session backend can be added behind the same SessionStore abstraction.

5. Normal Chat

from prashflow import Chat

chat = Chat(
    llm="ollama:qwen3:8b",
    session_id="user-001",
)

print(chat.chat("My name is Prash."))
print(chat.chat("What is my name?"))

Streaming:

for token in chat.stream("Explain Kubernetes"):
    print(token, end="", flush=True)

6. Agentic Ollama Chat

Use AgentChat when the model should decide when to call tools.

from prashflow import AgentChat

agent = AgentChat(
    llm="ollama:qwen3:8b",
    tools=["calculator"],
    session_id="user-001",
)

for token in agent.stream("Calculate 25% of 8000"):
    print(token, end="", flush=True)

PrashFlow uses LangGraph internally for the agent loop. The application does not need to build StateGraph or ToolNode itself.

7. Custom Python tools

from prashflow import AgentChat, tool

@tool
def get_server_status(server: str) -> str:
    """Return the status of a Linux server."""
    return f"{server}: UP"

agent = AgentChat(
    llm="ollama:qwen3:8b",
    tools=[get_server_status],
)

print(agent.run("Check web01"))

Tools can also require interactive approval:

@tool(requires_approval=True, max_retries=2)
def restart_service(server: str, service: str) -> str:
    """Restart a Linux service."""
    # implement the real operation here
    return f"Restarted {service} on {server}"

8. Multi-agent

Supervisor

from prashflow import MultiAgent

team = MultiAgent(
    model="ollama:qwen3:8b",
    mode="supervisor",
    agents=[
        {
            "name": "researcher",
            "description": "Research technical information.",
            "tools": ["web_search"],
        },
        {
            "name": "calculator",
            "description": "Perform arithmetic calculations.",
            "tools": ["calculator"],
        },
    ],
)

print(team.run("Calculate 20% of 8000"))

The supervisor chooses the specialist.

Sequential

team = MultiAgent(
    model="ollama:qwen3:8b",
    mode="sequential",
    agents=[
        {"name": "planner", "description": "Create a plan."},
        {"name": "developer", "description": "Develop the solution."},
        {"name": "reviewer", "description": "Review the solution."},
    ],
)

Flow:

Planner -> Developer -> Reviewer -> Final

Parallel

team = MultiAgent(
    model="ollama:qwen3:8b",
    mode="parallel",
    agents=[
        {"name": "security", "description": "Analyze security."},
        {"name": "performance", "description": "Analyze performance."},
        {"name": "architecture", "description": "Analyze architecture."},
    ],
)

The current reference implementation runs the specialist calls independently and synthesizes their results. An async concurrent implementation can be added for high-throughput production workloads.

Streaming multi-agent

for token in team.stream("Analyze this deployment"):
    print(token, end="", flush=True)

Per-agent models

team = MultiAgent(
    model="ollama:qwen3:8b",
    agents=[
        {
            "name": "researcher",
            "model": "ollama:qwen3:8b",
            "description": "Research information.",
        },
        {
            "name": "coder",
            "model": "ollama:qwen2.5-coder:14b",
            "description": "Write and review code.",
        },
    ],
)

9. RAG + Agent

from prashflow import AgentChat, RAG

rag = RAG(
    llm="ollama:qwen3:8b",
    embeddings="ollama:nomic-embed-text",
    vector_db="chroma:./data/chroma",
    retrieval={"type": "hybrid", "top_k": 5},
)
rag.ingest("./knowledge")

agent = AgentChat(
    llm="ollama:qwen3:8b",
    tools=[rag.as_tool(), "calculator"],
)

print(agent.run("Find our production deployment procedure."))

10. Multi-agent + RAG

team = MultiAgent(
    model="ollama:qwen3:8b",
    mode="supervisor",
    agents=[
        {
            "name": "company_knowledge",
            "description": "Answer questions from company documents.",
            "tools": [rag.as_tool()],
        },
        {
            "name": "calculator",
            "description": "Perform calculations.",
            "tools": ["calculator"],
        },
    ],
)

11. SQL ingestion

rag.ingest_sql(
    url="postgresql+psycopg://user:password@localhost:5432/company",
    query="SELECT id, title, description FROM incidents",
    content_columns=["title", "description"],
    metadata_columns=["id"],
)

MySQL is supported through the SQLAlchemy connection URL when the MySQL extra is installed.

12. MCP

PrashFlow provides an MCP configuration boundary so MCP can be attached to agents without changing the agent API.

agent = AgentChat(
    llm="ollama:qwen3:8b",
    mcp_servers=[
        {
            "name": "filesystem",
            "transport": "stdio",
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "./workspace"],
        }
    ],
)

The reference package validates MCP server configuration. For production MCP transport/session discovery, pin and implement against the MCP SDK version used by your organization; MCP SDK transport APIs can evolve.

13. LiteLLM

from prashflow import Chat

chat = Chat(
    llm={
        "provider": "litellm",
        "model": "openai/gpt-4.1",
    }
)

The application API remains chat.chat() / chat.stream() while the provider is selected by LiteLLM.

14. OpenAI / OpenAI-compatible

from prashflow import Chat

chat = Chat(
    llm={
        "provider": "openai-compatible",
        "model": "my-model",
        "base_url": "http://localhost:8000/v1",
        "api_key": "dummy",
    }
)

15. YAML configuration

prashflow.yaml:

llm:
  provider: ollama
  model: qwen3:8b
  base_url: http://localhost:11434

embeddings:
  provider: ollama
  model: nomic-embed-text
  base_url: http://localhost:11434

vector_db:
  provider: chroma
  path: ./data/chroma
  collection: company_docs

retrieval:
  type: hybrid
  top_k: 5
  candidate_k: 20
  semantic_weight: 0.6
  keyword_weight: 0.4

chunking:
  size: 1000
  overlap: 200

reranker:
  enabled: false

Load it:

from prashflow import RAG

rag = RAG.from_config("prashflow.yaml")

Environment variables are supported:

llm:
  provider: openai
  model: ${OPENAI_MODEL}
  api_key: ${OPENAI_API_KEY}

16. Error handling

PrashFlow exposes typed exceptions:

from prashflow import PrashFlowError

try:
    print(rag.ask("What is our deployment process?"))
except PrashFlowError as exc:
    print(f"PrashFlow error: {exc}")

Available categories include configuration, LLM, embedding, vector DB, document loading, retrieval, reranking, tool, agent and MCP errors.

17. Package structure

prashflow/
├── pyproject.toml
├── README.md
├── PRASHFLOW_GUIDE.md
├── examples/
├── tests/
└── src/prashflow/
    ├── __init__.py
    ├── __version__.py
    ├── agent.py
    ├── agentchat.py
    ├── chat.py
    ├── config.py
    ├── display.py
    ├── errors.py
    ├── inmemory.py
    ├── llm.py
    ├── loaders.py
    ├── mcp.py
    ├── memory.py
    ├── multiagent.py
    ├── rag.py
    ├── rerank.py
    ├── retrieval.py
    ├── session.py
    ├── sql_ingest.py
    ├── tools.py
    └── vectorstores.py

18. Build the pip package

python -m pip install --upgrade build
python -m build

Output:

dist/
├── prashflow-1.4.0-py3-none-any.whl
└── prashflow-1.4.0.tar.gz

Test the wheel:

pip install dist/prashflow-1.4.0-py3-none-any.whl
python -c "from prashflow import Chat, RAG, Agent, AgentChat, MultiAgent; print('PrashFlow OK')"

Publish when ready:

python -m pip install --upgrade twine
twine check dist/*
twine upload dist/*

19. Design philosophy

Application developers should write:

from prashflow import RAG, AgentChat, MultiAgent

and should not need to directly assemble LangChain loaders, LangGraph state graphs, Chroma clients, BM25 indexes, tool nodes or model-provider adapters for common use cases.

Advanced developers can still customize the underlying components when needed.

Roadmap

Planned production enhancements:

• true async/parallel specialist execution
• persistent Redis/PostgreSQL session backends
• complete MCP client/session discovery against a pinned SDK
• A2A support
• model fallback and cost routing
• observability/tracing
• structured output and Pydantic schemas
• FastAPI integration helpers
• ingestion manifests and changed-file detection
• background ingestion jobs
• citation objects with source/page metadata

PrashFlow 1.5 — Runtime Params, Token Usage, Reasoning and A2A

PrashFlow 1.5 keeps the existing src/prashflow/ flat structure. The existing RAG, Agent, AgentChat, MultiAgent, MCP, retrieval and vector-store APIs remain available.

Params

from prashflow import RAG, Params

params = Params(
    context_window=32768,
    max_context_tokens=12000,
    reserve_output_tokens=2000,
    thinking_level="high",
    top_k=8,
    search_type="mmr",
    max_input_tokens=12000,
    max_output_tokens=2000,
    max_total_tokens=14000,
    context_strategy="relevance",
    track_usage=True,
)

rag = RAG(
    llm={"provider": "ollama", "model": "qwen3:8b"},
    embeddings={"provider": "ollama", "model": "nomic-embed-text"},
    vector_db={"provider": "chroma", "path": "./chroma_db"},
    params=params,
)

Backward-compatible result API

Existing code remains:

answer = rag.ask("What is our deployment process?")

and returns a string.

For token/context/source metadata:

result = rag.ask_result("What is our deployment process?")

print(result.answer)
print(result.usage.input_tokens)
print(result.usage.output_tokens)
print(result.usage.total_tokens)
print(result.usage.context_tokens)
print(result.usage.utilization)
print(result.thinking_level)

Provider usage is preferred. If a provider does not return usage metadata, PrashFlow uses an estimate and marks result.usage.estimated == True.

Thinking levels

off
low
medium
high
auto

These are translated to provider-specific settings where supported. PrashFlow does not expose or store private chain-of-thought.

A2A

A2A is optional:

pip install "prashflow[a2a]"

Client:

from prashflow import A2AClient

agent = A2AClient(
    "http://localhost:9001/",
    name="security",
)

await agent.connect()
result = await agent.ask(
    "Analyze this architecture."
)
print(result.answer)
await agent.close()

Multiple remote agents:

from prashflow import A2ARegistry, A2AOrchestrator

registry = A2ARegistry({
    "security": "http://localhost:9001/",
    "devops": "http://localhost:9002/",
    "database": "http://localhost:9003/",
})

await registry.connect_all()

orchestrator = A2AOrchestrator(registry)

results = await orchestrator.broadcast(
    "Analyze this production architecture.",
    agents=["security", "devops", "database"],
)

Expose a PrashFlow RAG/Agent runtime:

from prashflow import run_a2a_server

run_a2a_server(
    rag,
    host="0.0.0.0",
    port=9999,
    name="PrashFlow RAG Agent",
)

A2A uses the official a2a-sdk Python client/server abstractions and Agent Card discovery.

PrashFlow 1.6 — Routing, Fallback, Health and RAG Evaluation

PrashFlow 1.6 preserves the existing flat src/prashflow/ structure and adds four production-oriented capabilities without changing the existing RAG, Agent, Chat, or MultiAgent entry points.

Model routing

A single model remains fully supported:

llm={"provider": "ollama", "model": "qwen3:8b"}

For multiple models:

llm={
    "strategy": "priority",
    "models": [
        {"provider": "ollama", "model": "qwen3:8b"},
        {"provider": "ollama", "model": "llama3.1:8b"},
        {"provider": "openai", "model": "gpt-5"},
    ],
    "fallback_on_error": True,
}

Supported routing strategies:

priority        first model first; fallback on failure
fallback        priority/fallback behavior
round_robin     rotate the starting model
balanced        prefer models with fewer recent failures
least_failures  same failure-aware policy
first_available priority with health checks

Automatic fallback

Fallback happens when the current provider/model raises an exception.

rag = RAG(
    llm={
        "strategy": "priority",
        "models": [
            {"provider": "ollama", "model": "qwen3:8b"},
            {"provider": "ollama", "model": "llama3.1:8b"},
        ],
    }
)

answer = rag.ask("Explain our deployment process.")

Existing single-model applications are unchanged.

Health check

status = rag.health_check()
print(status)

Deep check:

status = rag.health_check(deep=True)

The deep check verifies LLM endpoint availability, embeddings and vector-store access. The normal check is intentionally lightweight.

For a routed model directly:

from prashflow import ModelRouter

router = ModelRouter([
    {"provider": "ollama", "model": "qwen3:8b"},
    {"provider": "ollama", "model": "llama3.1:8b"},
])

print(router.health_check())

Health-aware routing

To check a model before every request:

params = Params(
    health_check_before_request=True,
    health_check_timeout=3,
)

rag = RAG(
    llm={
        "strategy": "priority",
        "models": [
            {"provider": "ollama", "model": "qwen3:8b"},
            {"provider": "ollama", "model": "llama3.1:8b"},
        ],
    },
    params=params,
)

This is more expensive than normal fallback, so it is opt-in.

RAG evaluation

PrashFlow includes a dependency-free regression evaluator:

evaluation = rag.evaluate([
    {
        "question": "Who approves production deployments?",
        "expected_answer": "The DevOps team approves production deployments.",
        "expected_sources": ["deployment.md"],
    },
    {
        "question": "What is the rollback procedure?",
        "expected_answer": "Rollback is performed using the previous release.",
    },
])

print(evaluation["averages"])

Metrics:

context_relevance
context_recall
answer_relevance
faithfulness
source_recall
overall

These are transparent lexical/coverage heuristics designed for regression testing. They are not a replacement for a semantic evaluator such as RAGAS or an LLM-as-a-judge system.