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 Agent
def get_server_status(server: str) -> str:
    """Get Linux server status."""
    return f"{server}: UP"
agent = Agent(
    llm={
        "provider": "ollama",
        "model": "qwen3:8b",
        "base_url": "http://localhost:11434",
    },
    tools=[
        "calculator",
        get_server_status,
    ],
)
print(
    agent.run(
        "Check web01 status and calculate 20 percent of 500."
    )
)

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. 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