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

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

P

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.

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.

PrashFlow — End User Guide

For users who install PrashFlow with pip and want to build AI/RAG applications.

You do not need to clone the PrashFlow repository or understand its internal files to use this guide.

---

1. What is PrashFlow?

PrashFlow is a Python framework that gives you a simple API for building:

• RAG applications
• LLM applications
• AI agents
• Multi-agent applications
• MCP integrations
• A2A agent integrations
• Model routing
• Automatic model fallback
• Context/token management
• Thinking/reasoning configuration
• Usage tracking
• Health checks
• RAG evaluation

The goal is to let you configure your AI infrastructure instead of writing the orchestration yourself.

Typical application:

Your Python Application
        |
        v
     PrashFlow
        |
   +----+----------------------+
   |    |          |           |
   v    v          v           v
  RAG  Agent      MCP         A2A
   |
   v
Model Router
   |
+--+-----------+-----------+
|              |           |
Ollama       OpenAI      LiteLLM
   |
   v
Vector Database
   |
   +--> Chroma
   +--> other supported stores

---

2. Install PrashFlow

You install PrashFlow like any normal Python package.

pip install prashflow

Verify:

python -c "import prashflow; print(prashflow.__version__)"

You should see the installed version.

---

3. Optional A2A Installation

If you want A2A support:

pip install "prashflow[a2a]"

If you only need RAG/LLM functionality, the base installation is sufficient.

---

4. Optional Provider Dependencies

PrashFlow uses provider integrations.

For example, if you want to use Ollama, install/run Ollama separately and pull the models you need.

Example:

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

Check:

ollama list

---

5. Your First PrashFlow Application

Create:

app.py

Add:

from prashflow import RAG

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

answer = rag.ask("What is PrashFlow?")

print(answer)

Run:

python app.py

That's the basic PrashFlow application.

---

6. What Do I Need to Configure?

For a basic RAG application, you normally provide:

LLM
Embedding Model
Vector Database
Prompt
Optional Params

Example:

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

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },
)

PrashFlow handles the orchestration.

---

7. LLM Configuration

Ollama

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

Custom Ollama endpoint:

llm={
    "provider": "ollama",
    "model": "qwen3:8b",
    "base_url": "http://localhost:11434",
}

For a remote Ollama server:

llm={
    "provider": "ollama",
    "model": "qwen3:8b",
    "base_url": "http://192.168.1.100:11434",
}

---

OpenAI

llm={
    "provider": "openai",
    "model": "gpt-5",
    "api_key": "YOUR_API_KEY",
}

For production applications, do not hard-code secrets.

Use environment variables or a secret manager.

---

LiteLLM

llm={
    "provider": "litellm",
    "model": "openai/gpt-5",
}

The exact model name and provider configuration depend on your LiteLLM setup.

---

8. Embedding Configuration

Embeddings convert your documents and user queries into vectors.

Example:

embeddings={
    "provider": "ollama",
    "model": "nomic-embed-text",
}

Conceptually:

Document
   |
   v
Embedding Model
   |
   v
Vector
   |
   v
Vector Database

Use a compatible and consistent embedding model when indexing and querying the same collection.

---

9. Vector Database

Chroma

The simplest local option:

vector_db={
    "provider": "chroma",
    "path": "./chroma_db",
}

This creates/uses a local Chroma database.

Your application can therefore look like:

my-ai-app/
├── app.py
└── chroma_db/

---

10. Complete Basic RAG Configuration

from prashflow import RAG

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

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },
)

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

print(answer)

---

11. Using Params

Advanced behavior is configured through Params.

from prashflow import RAG, Params

params = Params(
    context_window=32768,
    max_context_tokens=12000,
    reserve_output_tokens=2000,

    thinking_level="medium",

    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,
)

---

12. Params Explained

context_window

context_window=32768

The model's configured context capacity.

Example:

32,768 tokens

means the model can work with a context budget of approximately 32K tokens, subject to the actual provider/model capabilities.

---

max_context_tokens

max_context_tokens=12000

Maximum amount of retrieved context that PrashFlow should attempt to place into the prompt.

This prevents retrieval from consuming the entire model context.

---

reserve_output_tokens

reserve_output_tokens=2000

Reserves space for the generated response.

Conceptually:

Model Context
+-----------------------------------+
| System / user / history           |
|                                   |
| Retrieved Context                 |
| max_context_tokens                |
|                                   |
| Reserved Output                   |
| reserve_output_tokens             |
+-----------------------------------+

---

max_input_tokens

max_input_tokens=12000

Maximum input-token budget.

---

max_output_tokens

max_output_tokens=2000

Maximum output-token budget.

---

max_total_tokens

max_total_tokens=14000

Maximum combined input/output budget.

---

top_k

top_k=8

Number of documents/results retrieved.

Higher values can improve recall but may increase:

• latency
• context size
• token consumption
• irrelevant context

Start with:

top_k=5

or:

top_k=8

and evaluate.

---

search_type

Example:

search_type="mmr"

MMR can help balance relevance and diversity.

---

context_strategy

Example:

context_strategy="relevance"

Controls how PrashFlow manages retrieved context.

---

track_usage

track_usage=True

Enables token/usage metadata.

---

13. Thinking Levels

PrashFlow supports:

thinking_level="off"
thinking_level="low"
thinking_level="medium"
thinking_level="high"
thinking_level="auto"

---

Off

Params(
    thinking_level="off"
)

Use for simple tasks where reasoning overhead is unnecessary.

Examples:

classification
simple extraction
simple formatting
simple Q&A

---

Low

Params(
    thinking_level="low"
)

Good for normal questions and simple RAG.

---

Medium

Params(
    thinking_level="medium"
)

Good default for most applications.

---

High

Params(
    thinking_level="high"
)

Useful for:

complex reasoning
architecture analysis
debugging
multi-step questions
complex coding

Higher reasoning can increase latency and token consumption.

---

Auto

Params(
    thinking_level="auto"
)

Allows the model/provider integration to use its default behavior.

---

Important

Thinking level is a runtime policy.

PrashFlow does not expose private chain-of-thought.

You receive:

Final Answer
+
Usage Metadata
+
Model Metadata

not hidden reasoning.

Actual thinking support depends on the selected model/provider.

---

14. Token Usage

Instead of:

answer = rag.ask(...)

use:

result = rag.ask_result(
    "Explain our deployment architecture."
)

Then:

print(result.answer)

Usage:

print("Input:", result.usage.input_tokens)
print("Output:", result.usage.output_tokens)
print("Total:", result.usage.total_tokens)
print("Context:", result.usage.context_tokens)
print("Utilization:", result.usage.utilization)
print("Estimated:", result.usage.estimated)

---

15. ask() vs ask_result()

Existing/simple API

answer = rag.ask("What is PrashFlow?")

Returns:

str

Metadata API

result = rag.ask_result(
    "What is PrashFlow?"
)

Returns a result object containing:

answer
sources
usage
thinking_level
provider
model
context
metadata

This means existing applications can continue using:

rag.ask(...)

without changing their code.

---

16. Token Utilization

Suppose:

context_window=32768

and total usage is:

8192 tokens

The utilization is approximately:

25%

Use:

print(result.usage.utilization)

Provider-reported usage is preferred.

If the provider does not return usage information, PrashFlow can estimate usage.

Check:

print(result.usage.estimated)

---

17. Custom Prompt

You can provide a prompt:

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

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },

    prompt="""
    You are an enterprise knowledge assistant.

    Answer only using the retrieved context.

    If the answer is not present in the context,
    say that you do not know.

    Context:
    {context}

    Question:
    {question}

    History:
    {history}
    """,
)

Common placeholders:

{context}
{question}
{history}

Retrieved documents should be treated as data, not as system-level instructions.

---

18. Ingesting Documents

The normal RAG pipeline is:

Documents
   |
   v
Loader
   |
   v
Chunks
   |
   v
Embeddings
   |
   v
Chroma

Use the ingestion/loader APIs exposed by your installed PrashFlow version.

After ingestion, users can query:

answer = rag.ask(
    "What does the deployment document say?"
)

---

19. Retrieval Settings

Example:

params = Params(
    top_k=8,
    search_type="mmr",
)

Tune retrieval based on your data.

A common starting point:

Params(
    top_k=5,
    search_type="mmr",
)

Then compare results using the evaluation functionality.

---

20. Streaming

For applications where you want output as it is generated:

for chunk in rag.ask_stream(
    "Explain Kubernetes architecture."
):
    print(chunk, end="", flush=True)

Useful for:

• chat UIs
• web applications
• long responses
• interactive assistants

---

21. Model Routing

You can configure multiple LLMs.

rag = RAG(
    llm={
        "strategy": "priority",

        "models": [
            {
                "provider": "ollama",
                "model": "qwen3:8b",
            },

            {
                "provider": "ollama",
                "model": "llama3.1:8b",
            },

            {
                "provider": "openai",
                "model": "gpt-5",
                "api_key": "YOUR_API_KEY",
            },
        ],

        "fallback_on_error": True,
    },

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },
)

---

22. Routing Strategies

Priority

strategy="priority"

Models are tried in the order provided.

Model 1
  |
  X
  |
Model 2
  |
  X
  |
Model 3

---

Fallback

strategy="fallback"

Uses priority/fallback behavior.

---

Round Robin

strategy="round_robin"

Rotates the starting model.

Useful when multiple models are equivalent and you want to distribute requests.

---

Balanced

strategy="balanced"

Prefers models with fewer recent failures.

---

Least Failures

strategy="least_failures"

Uses recent failure counts when deciding which model to use.

---

First Available

strategy="first_available"

Useful with health-aware routing.

---

23. Automatic Fallback

Example:

llm={
    "strategy": "priority",

    "models": [
        {
            "provider": "ollama",
            "model": "qwen3:8b",
        },

        {
            "provider": "ollama",
            "model": "llama3.1:8b",
        },
    ],

    "fallback_on_error": True,
}

If the first model fails:

qwen3:8b
     |
     X
     |
     v
llama3.1:8b
     |
     v
Response

You can disable fallback:

"fallback_on_error": False

---

24. Testing Fallback

For testing, intentionally configure an invalid first model:

llm={
    "strategy": "priority",

    "models": [
        {
            "provider": "ollama",
            "model": "model-that-does-not-exist",
        },

        {
            "provider": "ollama",
            "model": "qwen3:8b",
        },
    ],

    "fallback_on_error": True,
}

Then:

answer = rag.ask(
    "Explain PrashFlow."
)

print(answer)

Expected:

Invalid model
     |
   error
     |
     v
qwen3:8b
     |
     v
answer

---

25. Health Check

Basic:

status = rag.health_check()

print(status)

Deep:

status = rag.health_check(
    deep=True
)

print(status)

Deep checking verifies the important RAG components:

LLM
 |
Embeddings
 |
Vector DB

Example:

{
    "healthy": True,
    "llm": {
        "healthy": True,
    },
    "embeddings": {
        "healthy": True,
    },
    "vector_db": {
        "healthy": True,
    },
}

---

26. Model Router Health Check

You can also directly use:

from prashflow import ModelRouter

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

status = router.health_check()

print(status)

The result contains individual model health information.

---

27. Health Check Before Every Request

You can enable:

from prashflow import Params

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

Then:

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

This performs an additional health operation before selecting a model.

For high-throughput applications, explicit health endpoints plus fallback are generally more efficient than checking on every request.

---

28. RAG Evaluation

PrashFlow includes a built-in evaluator for RAG regression testing.

Example:

evaluation = rag.evaluate([
    {
        "question": "Who approves production deployments?",

        "expected_answer": (
            "The DevOps team approves production deployments."
        ),

        "expected_sources": [
            "deployment.md"
        ],
    },

    {
        "question": "What is used for CI/CD?",

        "expected_answer": (
            "Jenkins is used for CI/CD."
        ),
    },
])

Print:

print(evaluation["averages"])

---

29. Evaluation Metrics

PrashFlow reports:

context_relevance
context_recall
answer_relevance
faithfulness
source_recall
overall

Example:

{
    "context_relevance": 0.91,
    "context_recall": 0.88,
    "answer_relevance": 0.94,
    "faithfulness": 0.90,
    "source_recall": 1.0,
    "overall": 0.926,
}

Per-question:

for item in evaluation["results"]:
    print(item["question"])
    print(item["metrics"])

These are transparent, dependency-free lexical/coverage heuristics.

They are useful for:

• regression tests
• comparing retrieval settings
• CI validation
• tuning RAG

They are not a complete semantic evaluation system.

---

30. Compare Two RAG Configurations

For example, test:

top_k=5

versus:

top_k=10

Run the same evaluation dataset.

Example:

Configuration A

Overall: 0.84


Configuration B

Overall: 0.91

This gives you a practical way to measure RAG changes.

---

31. Agents

Create an agent:

from prashflow import Agent

agent = Agent(
    llm={
        "provider": "ollama",
        "model": "qwen3:8b",
    },
)

Then use the existing Agent API:

answer = agent.run(
    "Explain Kubernetes deployments."
)

print(answer)

Agents are useful when you need:

LLM
+
Tools
+
Decision making
+
Memory

---

32. AgentChat

from prashflow import AgentChat

chat = AgentChat(
    llm={
        "provider": "ollama",
        "model": "qwen3:8b",
    },
)

Use the conversational methods exposed by your installed PrashFlow version.

---

33. Multi-Agent

PrashFlow supports multi-agent applications.

Concept:

                 User Request
                      |
                      v
                Multi-Agent
                      |
          +-----------+-----------+
          |           |           |
          v           v           v
       Security     DevOps     Database
        Agent        Agent       Agent
          |           |           |
          +-----------+-----------+
                      |
                      v
                 Final Result

Use the MultiAgent API included in your installed version.

---

34. Tools

PrashFlow provides tools:

from prashflow import tool, ToolManager

Tools allow agents to interact with external functionality.

Examples:

Python functions
REST APIs
Databases
MCP tools
RAG
A2A agents

---

35. MCP

PrashFlow supports MCP.

Conceptually:

PrashFlow Agent
      |
      v
   MCP Client
      |
  +---+---+---+
  |   |   |   |
 Git DB  API Files

Use the MCP APIs available in your installed version.

---

36. A2A

A2A allows PrashFlow agents to communicate with remote agents.

Install:

pip install "prashflow[a2a]"

Example:

from prashflow import A2AClient

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

Connect:

await agent.connect()

Send a request:

result = await agent.ask(
    "Analyze this architecture for security risks."
)

print(result.answer)

Close:

await agent.close()

---

37. Multiple A2A Agents

Create a registry:

from prashflow import A2ARegistry

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

Connect:

await registry.connect_all()

List:

print(registry.names())

---

38. A2A Skill Routing

Search for agents by skill:

matches = registry.find_by_skill(
    "security"
)

Use A2ARouter:

from prashflow import A2ARouter

router = A2ARouter(registry)

result = await router.ask(
    "Check this deployment for security risks.",
    skill="security",
)

print(result.answer)

Explicit agent:

result = await router.ask(
    "Analyze this deployment.",
    agent="security",
)

---

39. A2A Orchestration

Use multiple remote agents:

from prashflow import A2AOrchestrator

orchestrator = A2AOrchestrator(registry)

results = await orchestrator.broadcast(
    "Analyze this production architecture.",

    agents=[
        "security",
        "devops",
        "database",
    ],
)

Flow:

                  Request
                     |
                     v
              A2A Orchestrator
                     |
        +------------+------------+
        |            |            |
        v            v            v
    Security       DevOps      Database
      Agent         Agent        Agent
        |            |            |
        +------------+------------+
                     |
                     v
                  Results

---

40. A2A Server

A PrashFlow runtime can be exposed as an A2A agent.

from prashflow import RAG, Params, run_a2a_server

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

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },

    params=Params(
        thinking_level="medium",
        track_usage=True,
    ),
)

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

---

41. Recommended Production Configuration

from prashflow import RAG, Params

params = Params(
    context_window=32768,

    max_context_tokens=12000,

    reserve_output_tokens=2000,

    thinking_level="medium",

    top_k=8,

    search_type="mmr",

    max_input_tokens=12000,

    max_output_tokens=2000,

    max_total_tokens=14000,

    context_strategy="relevance",

    track_usage=True,

    fallback_on_error=True,

    health_check_timeout=5,
)

rag = RAG(
    llm={
        "strategy": "priority",

        "models": [
            {
                "provider": "ollama",
                "model": "qwen3:8b",
            },

            {
                "provider": "ollama",
                "model": "llama3.1:8b",
            },
        ],

        "fallback_on_error": True,
    },

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },

    params=params,
)

---

42. Complete End-User Example

Create:

app.py

from prashflow import RAG, Params


params = Params(
    context_window=32768,

    max_context_tokens=12000,

    reserve_output_tokens=2000,

    thinking_level="medium",

    top_k=8,

    search_type="mmr",

    max_input_tokens=12000,

    max_output_tokens=2000,

    max_total_tokens=14000,

    context_strategy="relevance",

    track_usage=True,

    fallback_on_error=True,
)


rag = RAG(
    llm={
        "strategy": "priority",

        "models": [
            {
                "provider": "ollama",
                "model": "qwen3:8b",
            },

            {
                "provider": "ollama",
                "model": "llama3.1:8b",
            },
        ],

        "fallback_on_error": True,
    },

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },

    params=params,
)


# -------------------------
# Health
# -------------------------

health = rag.health_check()

print("HEALTH")
print(health)


# -------------------------
# Ask
# -------------------------

result = rag.ask_result(
    "Explain our production deployment process."
)


print("\nANSWER")
print(result.answer)


# -------------------------
# Usage
# -------------------------

if result.usage:

    print("\nUSAGE")

    print(
        "Input tokens:",
        result.usage.input_tokens,
    )

    print(
        "Output tokens:",
        result.usage.output_tokens,
    )

    print(
        "Total tokens:",
        result.usage.total_tokens,
    )

    print(
        "Context tokens:",
        result.usage.context_tokens,
    )

    print(
        "Utilization:",
        result.usage.utilization,
    )

    print(
        "Estimated:",
        result.usage.estimated,
    )


# -------------------------
# Thinking
# -------------------------

print("\nTHINKING LEVEL")
print(result.thinking_level)


# -------------------------
# Sources
# -------------------------

print("\nSOURCES")

for source in result.sources:
    print(source)


# -------------------------
# Evaluation
# -------------------------

evaluation = rag.evaluate([
    {
        "question": (
            "Explain our production deployment process."
        ),

        "expected_answer": (
            "Production deployment is approved "
            "and executed through the defined "
            "deployment process."
        ),
    }
])


print("\nEVALUATION")

print(
    evaluation["averages"]
)

Run:

python app.py

---

43. Minimal Application

If you don't need advanced features, you only need:

from prashflow import RAG

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

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },
)

print(
    rag.ask("What is PrashFlow?")
)

You can start here and add advanced features later.

---

44. Feature Summary

After installing:

pip install prashflow

you can build:

PrashFlow
|
+-- RAG
|   +-- Documents
|   +-- Embeddings
|   +-- Vector DB
|   +-- Retrieval
|   +-- MMR
|   +-- Reranking
|   +-- Context
|   +-- Prompts
|   +-- Sessions
|
+-- LLM
|   +-- Ollama
|   +-- OpenAI
|   +-- LiteLLM
|   +-- Model Routing
|   +-- Automatic Fallback
|
+-- Runtime
|   +-- Context Window
|   +-- Token Usage
|   +-- Token Utilization
|   +-- Thinking Levels
|   +-- Health Checks
|
+-- Evaluation
|   +-- Context Relevance
|   +-- Context Recall
|   +-- Answer Relevance
|   +-- Faithfulness
|   +-- Source Recall
|   +-- Overall Score
|
+-- Agents
|   +-- Agent
|   +-- AgentChat
|   +-- Multi-Agent
|   +-- Tools
|
+-- Protocols
    +-- MCP
    +-- A2A
        +-- Client
        +-- Registry
        +-- Router
        +-- Orchestrator
        +-- Server

---

45. Recommended Learning Path

If you are new to PrashFlow:

Step 1 — Basic RAG

rag = RAG(...)
rag.ask(...)

Step 2 — Add Params

Params(
    top_k=5,
    thinking_level="medium",
)

Step 3 — Add usage tracking

rag.ask_result(...)

Step 4 — Add model fallback

llm={
    "strategy": "priority",
    "models": [...]
}

Step 5 — Add health checks

rag.health_check()

Step 6 — Evaluate RAG

rag.evaluate([...])

Step 7 — Build agents

Agent(...)

Step 8 — Add MCP

Connect external tools.

Step 9 — Add A2A

Connect remote agents.

Step 10 — Build your production AI application

RAG
+
Agents
+
Tools
+
MCP
+
A2A
+
Model Routing
+
Fallback
+
Usage
+
Evaluation
+
Health

---

46. Important End-User Rule

You normally do not need to import or modify PrashFlow's internal modules.

Prefer:

from prashflow import RAG, Params

instead of importing internal implementation files.

Your application should look like:

my-ai-app/
|
+-- app.py
+-- documents/
+-- chroma_db/
+-- .env
└-- requirements.txt

and PrashFlow should be installed as:

pip install prashflow

The framework handles the underlying orchestration.

---

47. Final Quick Start

Install:

pip install prashflow

Install Ollama models if using Ollama:

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

Create:

from prashflow import RAG, Params

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

    embeddings={
        "provider": "ollama",
        "model": "nomic-embed-text",
    },

    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
    },

    params=Params(
        thinking_level="medium",
        top_k=8,
        track_usage=True,
    ),
)

result = rag.ask_result(
    "What is PrashFlow?"
)

print(result.answer)

if result.usage:
    print(
        "Total tokens:",
        result.usage.total_tokens,
    )

"""
PrashFlow 1.2 - Production-style Client RAG Application

Features:
    - Document ingestion
    - Persistent Chroma vector DB
    - Hybrid semantic + BM25 retrieval
    - Optional reranking
    - Model routing / fallback
    - Streaming RAG chat
    - Session memory
    - Source display
    - Usage statistics
    - Health checks
    - RAG evaluation
    - Interactive CLI

Install:
    pip install "prashflow[all]"

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

Run:
    python app.py
"""

from pathlib import Path

from prashflow import RAG, Params


# ============================================================
# APPLICATION CONFIGURATION
# ============================================================

APP_NAME = "PrashFlow Enterprise RAG"

KNOWLEDGE_DIR = "./knowledge"
VECTOR_DIR = "./data/chroma"

COLLECTION_NAME = "company_knowledge"

SESSION_ID = "user-001"


# ============================================================
# MODEL CONFIGURATION
# ============================================================

# Primary + fallback models.
#
# If you only have one model, simply keep one entry.
#
# PrashFlow will try the first model and fall back to
# the next model if the first one fails.

LLM_CONFIG = {
    "strategy": "priority",

    "models": [
        {
            "provider": "ollama",
            "model": "qwen3:8b",
            "base_url": "http://localhost:11434",
        },

        {
            "provider": "ollama",
            "model": "llama3.1:8b",
            "base_url": "http://localhost:11434",
        },
    ],

    "fallback_on_error": True,
}


# Embedding model.
#
# Keep embeddings stable once your knowledge base is indexed.
EMBEDDING_CONFIG = {
    "provider": "ollama",
    "model": "nomic-embed-text",
    "base_url": "http://localhost:11434",
}


# ============================================================
# PRASHFLOW PARAMETERS
# ============================================================

PARAMS = Params(
    # Model/context controls
    context_window=32768,

    # Don't send unnecessarily huge context to the LLM.
    max_context_tokens=6000,

    # Reserve output space.
    reserve_output_tokens=1200,

    # Retrieval defaults
    top_k=4,
    search_type="hybrid",

    # Usage tracking
    track_usage=True,

    # Resilience
    fallback_on_error=True,

    # IMPORTANT:
    # Don't perform a health request before EVERY query.
    # That adds latency.
    health_check_before_request=False,

    health_check_timeout=5.0,

    # Evaluation disabled during normal user requests.
    evaluation_enabled=False,
)


# ============================================================
# RAG ENGINE
# ============================================================

rag = RAG(

    # --------------------------------------------------------
    # LLM / MODEL ROUTING
    # --------------------------------------------------------

    llm=LLM_CONFIG,

    # --------------------------------------------------------
    # EMBEDDINGS
    # --------------------------------------------------------

    embeddings=EMBEDDING_CONFIG,

    # --------------------------------------------------------
    # VECTOR DATABASE
    # --------------------------------------------------------

    vector_db={
        "provider": "chroma",

        "path": VECTOR_DIR,

        "collection": COLLECTION_NAME,
    },

    # --------------------------------------------------------
    # RETRIEVAL
    # --------------------------------------------------------

    retrieval={
        # Best general-purpose default.
        "type": "hybrid",

        # Final documents sent to the LLM.
        "top_k": 4,

        # Candidates retrieved before fusion.
        #
        # Keep this reasonably small for latency.
        "candidate_k": 12,

        # Hybrid weighting.
        "semantic_weight": 0.65,
        "keyword_weight": 0.35,
    },

    # --------------------------------------------------------
    # RERANKER
    # --------------------------------------------------------

    # Disabled for maximum speed.
    #
    # Enable when retrieval precision is more important
    # than latency.
    reranker={
        "enabled": False,
    },

    # --------------------------------------------------------
    # CHUNKING
    # --------------------------------------------------------

    chunking={
        "size": 900,
        "overlap": 120,
    },

    # --------------------------------------------------------
    # PRASHFLOW PARAMS
    # --------------------------------------------------------

    params=PARAMS,

    # Default session.
    session_id=SESSION_ID,
)


# ============================================================
# INGESTION
# ============================================================

def ingest_knowledge():
    """
    Index all documents under ./knowledge.

    Example:

        knowledge/
        ├── architecture.pdf
        ├── deployment.docx
        ├── security.md
        ├── troubleshooting.txt
        └── faq.csv
    """

    knowledge = Path(KNOWLEDGE_DIR)

    if not knowledge.exists():
        knowledge.mkdir(
            parents=True,
            exist_ok=True,
        )

        print(
            f"\nCreated knowledge directory: {knowledge}"
        )

        print(
            "Put your documents inside it and run again."
        )

        return

    supported = (
        ".pdf",
        ".docx",
        ".txt",
        ".md",
        ".csv",
    )

    files = [
        file
        for file in knowledge.rglob("*")
        if file.is_file()
        and file.suffix.lower() in supported
    ]

    if not files:
        print("\nNo supported documents found.")

        print(
            "Supported: PDF, DOCX, TXT, MD, CSV"
        )

        return

    print(
        f"\nFound {len(files)} documents."
    )

    print("Starting ingestion...\n")

    # PrashFlow handles loading, chunking,
    # embeddings and vector storage.
    #
    # Calling ingest on the directory keeps
    # application code extremely small.
    rag.ingest(KNOWLEDGE_DIR)

    print("\nKnowledge base ready.")


# ============================================================
# HEALTH CHECK
# ============================================================

def health_check():
    """
    Run a deep PrashFlow health check.

    This checks:
        - LLM
        - embeddings
        - vector DB
    """

    print("\nRunning PrashFlow health check...\n")

    result = rag.health_check(
        deep=True
    )

    print(
        f"Overall: "
        f"{'HEALTHY' if result['healthy'] else 'UNHEALTHY'}"
    )

    print(
        f"LLM: {result.get('llm')}"
    )

    print(
        f"Embeddings: {result.get('embeddings')}"
    )

    print(
        f"Vector DB: {result.get('vector_db')}"
    )

    return result


# ============================================================
# ONE-SHOT RAG
# ============================================================

def ask_once(question: str):
    """
    Non-streaming RAG.

    Useful for scripts, APIs and automation.
    """

    result = rag.ask_result(
        question,

        search_type="hybrid",

        top_k=4,

        candidate_k=12,

        # Keep reranking disabled for speed.
        rerank=False,
    )

    print("\n" + "=" * 70)

    print("ANSWER")

    print("=" * 70)

    print(result.answer)

    print("\n" + "=" * 70)

    print("SOURCES")

    print("=" * 70)

    for source in result.sources:

        print(
            f"- {source.get('source')}"
        )

    if result.usage:

        print("\n" + "=" * 70)

        print("USAGE")

        print("=" * 70)

        print(
            f"Input tokens : "
            f"{result.usage.input_tokens}"
        )

        print(
            f"Output tokens: "
            f"{result.usage.output_tokens}"
        )

        print(
            f"Total tokens : "
            f"{result.usage.total_tokens}"
        )

    return result


# ============================================================
# STREAMING RAG CHAT
# ============================================================

def chat():
    """
    Interactive conversational RAG.

    Uses PrashFlow chat_stream().

    Conversation history is maintained using
    the PrashFlow session.
    """

    print("\n" + "=" * 70)

    print(APP_NAME)

    print("=" * 70)

    print(
        "\nType your question."
    )

    print(
        "Commands:"
    )

    print(
        "  /exit     Exit"
    )

    print(
        "  /health   Health check"
    )

    print(
        "  /clear    Clear conversation"
    )

    print(
        "  /search   Change retrieval mode"
    )

    print()

    search_type = "hybrid"

    while True:

        try:

            question = input("You: ").strip()

        except (
            KeyboardInterrupt,
            EOFError,
        ):

            print("\nGoodbye.")

            break

        if not question:
            continue

        # ----------------------------------------------------
        # EXIT
        # ----------------------------------------------------

        if question.lower() in {
            "/exit",
            "exit",
            "quit",
        }:

            print("Goodbye.")

            break

        # ----------------------------------------------------
        # HEALTH
        # ----------------------------------------------------

        if question.lower() == "/health":

            health_check()

            continue

        # ----------------------------------------------------
        # CLEAR SESSION
        # ----------------------------------------------------

        if question.lower() == "/clear":

            rag.clear_session(
                SESSION_ID
            )

            print(
                "Conversation cleared."
            )

            continue

        # ----------------------------------------------------
        # SEARCH MODE
        # ----------------------------------------------------

        if question.lower() == "/search":

            print(
                "\nChoose search type:"
            )

            print(
                "1. hybrid"
            )

            print(
                "2. semantic"
            )

            print(
                "3. keyword"
            )

            print(
                "4. mmr"
            )

            print(
                "5. multi_query"
            )

            choice = input(
                "\nChoice: "
            ).strip()

            modes = {
                "1": "hybrid",
                "2": "semantic",
                "3": "keyword",
                "4": "mmr",
                "5": "multi_query",
            }

            search_type = modes.get(
                choice,
                "hybrid",
            )

            print(
                f"Search mode: {search_type}"
            )

            continue

        # ----------------------------------------------------
        # STREAMING RAG
        # ----------------------------------------------------

        print("\nAI: ", end="", flush=True)

        try:

            for token in rag.chat_stream(

                query=question,

                session_id=SESSION_ID,

                search_type=search_type,

                top_k=4,

                candidate_k=12,

                # Maximum speed.
                rerank=False,
            ):

                print(
                    token,
                    end="",
                    flush=True,
                )

            print("\n")

            # ------------------------------------------------
            # USAGE
            # ------------------------------------------------

            if rag.last_usage:

                usage = rag.last_usage

                print(
                    f"[tokens: "
                    f"{usage.total_tokens}]"
                )

                print()

        except Exception as exc:

            print(
                f"\nRAG error: {exc}\n"
            )


# ============================================================
# EVALUATION
# ============================================================

def evaluate():

    """
    Run a small RAG evaluation set.

    This should normally be used during development/CI,
    not on every user request.
    """

    test_cases = [

        {
            "question":
                "What is the production deployment process?",

            "expected_answer":
                "Production deployment requires approval "
                "and validation.",

            "expected_sources": [
                "deployment.pdf"
            ],
        },

        {
            "question":
                "How do I troubleshoot a failed deployment?",

            "expected_answer":
                "Check the deployment logs and "
                "validate the failed stage.",
        },
    ]

    print(
        "\nRunning RAG evaluation...\n"
    )

    result = rag.evaluate(
        test_cases
    )

    print(result)

    return result


# ============================================================
# APPLICATION MENU
# ============================================================

def main():

    print("\n" + "=" * 70)

    print(APP_NAME)

    print("=" * 70)

    print(
        "\n1. Ingest knowledge"
    )

    print(
        "2. Health check"
    )

    print(
        "3. Chat"
    )

    print(
        "4. Ask one question"
    )

    print(
        "5. Evaluate RAG"
    )

    print(
        "6. Exit"
    )

    while True:

        choice = input(
            "\nSelect: "
        ).strip()

        if choice == "1":

            ingest_knowledge()

        elif choice == "2":

            health_check()

        elif choice == "3":

            chat()

        elif choice == "4":

            question = input(
                "\nQuestion: "
            ).strip()

            if question:

                ask_once(question)

        elif choice == "5":

            evaluate()

        elif choice == "6":

            print(
                "Goodbye."
            )

            break

        else:

            print(
                "Invalid option."
            )


# ============================================================
# ENTRY POINT
# ============================================================

if __name__ == "__main__":

    main()

That's all you need to get started.

Install PrashFlow → configure your LLM → configure embeddings → configure vector DB → build your AI application.

PrashFlow 1.3 — Guardrails, Human-in-the-Loop & Incremental RAG

PrashFlow 1.3 keeps the existing v1.2 APIs and file layout intact while adding opt-in production safety and efficient RAG synchronization.

Model-agnostic guardrails

Guardrails can use deterministic rules, custom Python functions, a dedicated security/classifier model, an LLM judge, or an external HTTP guardrail service. The guard model is independent of the application's generation model.

from prashflow import RAG

rag = RAG(
    llm={"provider": "ollama", "model": "qwen3:8b"},
    embeddings={"provider": "ollama", "model": "nomic-embed-text"},
    vector_db={
        "provider": "chroma",
        "path": "./chroma_db",
        "collection": "company_docs",
    },
    guardrails={
        "enabled": True,
        "prompt_injection": True,
        "pii": {"action": "redact"},
        "input": {"max_length": 8000},
        "grounding": {"action": "warn", "threshold": 0.15},
    },
)

Supported guard decisions:

allow, deny, block, warn, redact, retry, approval.

A model-backed guard can be configured independently:

guardrails={
    "enabled": True,
    "prompt_injection": {
        "model": {
            "provider": "ollama",
            "model": "llama-guard",
        }
    },
}

Custom guards are also supported:

def my_policy(stage, value, metadata):
    if stage == "tool" and metadata.get("tool") == "delete_production":
        return {"decision": "deny", "risk": "critical"}
    return {"decision": "allow"}

guardrails = {
    "enabled": True,
    "custom": [my_policy],
}

Tool security

from prashflow import Agent

agent = Agent(
    llm="ollama:qwen3:8b",
    tools=[database_tool, deploy_tool],
    guardrails={
        "enabled": True,
        "tools": {
            "database_tool": {"risk": "low", "action": "allow"},
            "deploy_tool": {"risk": "high", "action": "approval"},
            "delete_database": {"risk": "critical", "action": "deny"},
        },
    },
    human_in_loop={
        "enabled": True,
        "provider": "cli",
    },
)

Human-in-the-loop

The same approval engine supports four interfaces:

• cli — local/development applications
• callback — custom Python applications
• api — web/mobile/custom UI
• streamlit — quick internal AI applications

CLI:

human_in_loop={"enabled": True, "provider": "cli"}

Callback:

def approve(request):
    return input(f"Approve {request.action}? [y/N] ").lower() == "y"

human_in_loop={
    "enabled": True,
    "provider": "callback",
    "callback": approve,
}

REST API:

from prashflow import ApprovalManager

manager = ApprovalManager(
    provider="api",
    auto_start_api=True,
    api_host="127.0.0.1",
    api_port=8765,
)

Endpoints:

GET  /approvals
GET  /approvals/{approval_id}
POST /approvals/{approval_id}/approve
POST /approvals/{approval_id}/reject

Install REST/Streamlit support with:

pip install "prashflow[hitl]"

Streamlit:

from prashflow import ApprovalManager

manager = ApprovalManager(provider="streamlit")
manager.streamlit()

The four frontends share the same ApprovalRequest and approval state. Agents pause while an API/Streamlit approval is pending and resume after approval or terminate after rejection/expiry.

Incremental RAG

Enable incremental indexing without changing the existing rag.ingest() behavior:

rag = RAG(
    llm={...},
    embeddings={...},
    vector_db={...},
    incremental={
        "enabled": True,
        "manifest": "./.prashflow/index_manifest.json",
        "bm25": "./.prashflow/bm25.json",
    },
)

result = rag.sync("./docs")
print(result)

Example result:

{
    "scanned": 1250,
    "new": 12,
    "modified": 7,
    "deleted": 3,
    "unchanged": 1228,
    "chunks_indexed": 184,
    "chunks_removed": 41
}

Only new/modified documents are embedded. Deleted documents have their old vectors removed. The BM25 sidecar is persisted so hybrid retrieval survives a process restart when incremental mode is enabled.

The existing:

rag.ingest("./docs")

remains unchanged.

Important compatibility rule

All v1.3 capabilities are opt-in. Existing v1.0/v1.2 code does not need to be rewritten and existing RAG, Agent, MCP, A2A, routing, fallback, streaming, evaluation and vector-store APIs remain available through:

from prashflow import RAG, Agent, Params