How to Configure Knowledge Filters with KnowledgeFilterService in OpenRAG

Use the KnowledgeFilterService class in src/services/knowledge_filter_service.py to create, search, update, and delete knowledge filters stored in the OpenSearch knowledge_filters index, while managing webhook subscriptions and alerting monitors through dedicated async methods.

OpenRAG (langflow-ai/openrag) persists knowledge filters as lightweight documents in an OpenSearch index. The KnowledgeFilterService acts as the single point of truth for all CRUD-style operations, subscription handling, and integration with the alerting monitor system. Understanding how to configure these filters correctly ensures your RAG pipelines respond to relevant document changes in real time.

Core Architecture and Data Model

OpenRAG initializes the knowledge_filters index at startup in src/main.py (lines 86‑104). Each filter document contains metadata fields including name, description, query_data, owner, allowed_users, allowed_groups, and a subscriptions array for webhook alerts.

The KnowledgeFilterService wraps the OpenSearch client to provide async methods for document indexing, searching, updating, and deletion. It enforces ownership and ACL checks on every operation. For real-time alerting, the service integrates with MonitorService in src/services/monitor_service.py, which creates OpenSearch Alerting plugin monitors that POST match results to generated webhook URLs.

Creating a Knowledge Filter

To store a new filter, instantiate the service with a valid session manager and call create_knowledge_filter. The method indexes the document into knowledge_filters and immediately refreshes the index for read-after-write consistency (lines 20‑34).

from src.services.knowledge_filter_service import KnowledgeFilterService

new_filter = {
    "id": "kf-1234",
    "name": "Recent‑AI‑Papers",
    "description": "All documents containing AI in the title from the last 30 days",
    "query_data": "title:AI AND timestamp:[now-30d TO now]",
    "owner": "user-42",
    "allowed_users": ["user-42", "user-99"],
    "allowed_groups": ["researchers"],
    "subscriptions": [],
    "created_at": "2026-03-13T12:00:00Z",
    "updated_at": "2026-03-13T12:00:00Z",
}

service = KnowledgeFilterService(session_manager)
result = await service.create_knowledge_filter(new_filter, user_id="user-42")

The method returns a dictionary containing success, id, and the stored filter object.

Searching and Retrieving Filters

The search_knowledge_filters method executes a free-text query across the name, description, and query_data fields. It builds a multi_match query with fuzziness and sorts results by relevance score then updated_at (lines 52‑66).

filters = await service.search_knowledge_filters(
    query="AI papers", user_id="user-42", limit=10
)

For direct retrieval by ID, use get_knowledge_filter (lines 10‑27), which validates ownership or ACL membership before returning the document.

Updating and Deleting Filters

Update operations merge partial changes into existing documents. The update_knowledge_filter method issues an OpenSearch update request, refreshes the index, and returns the updated document (lines 48‑66).

updates = {
    "description": "All AI‑related papers from the last 60 days",
    "updated_at": "2026-03-13T13:15:00Z",
}
result = await service.update_knowledge_filter(
    filter_id="kf-1234", updates=updates, user_id="user-42"
)

Deletion is immediate and permanent. Call delete_knowledge_filter with the filter ID and requesting user ID; the method sends a delete request and refreshes the index to prevent stale reads (lines 82‑94).

await service.delete_knowledge_filter(filter_id="kf-1234", user_id="user-42")

Managing Webhook Subscriptions

Knowledge filters support webhook subscriptions that trigger external systems when the filter query matches new documents. The service stores subscriptions as objects inside the filter document’s subscriptions array.

Adding a Subscription

The add_subscription method appends the payload to the subscriptions array and updates the document (lines 41‑66).

subscription = {
    "subscription_id": "sub-001",
    "created_at": "2026-03-13T14:00:00Z",
    "notify_email": "[email protected]"
}
await service.add_subscription(
    filter_id="kf-1234",
    subscription_data=subscription,
    user_id="user-42"
)

Listing Subscriptions

Retrieve the current subscription array without modification using get_filter_subscriptions (lines 125‑138).

subs = await service.get_filter_subscriptions("kf-1234", user_id="user-42")

Removing a Subscription

The remove_subscription method rebuilds the subscriptions array excluding the matching subscription_id (lines 70‑103).

await service.remove_subscription(
    filter_id="kf-1234",
    subscription_id="sub-001",
    user_id="user-42"
)

Integrating with OpenSearch Alert Monitors

While KnowledgeFilterService manages the filter document, the MonitorService class creates the actual alerting infrastructure. It converts the filter’s query_data into an OpenSearch Alerting DSL and registers a document-level monitor that POSTs results to a generated webhook URL (lines 16‑124).

from src.services.monitor_service import MonitorService

monitor_srv = MonitorService(session_manager, webhook_base_url="https://my-openrag/api")
monitor = await monitor_srv.create_knowledge_filter_monitor(
    filter_id="kf-1234",
    filter_name="Recent‑AI‑Papers",
    query_data={"query": {"match": {"title": "AI"}}},
    user_id="user-42",
    jwt_token="eyJhbGci...",
    notification_config={"email": "[email protected]"}
)

This generates a webhook endpoint at /knowledge-filter/{filter_id}/webhook/{subscription_id} that the alerting plugin calls when matches occur.

REST API Exposure

All service methods are exposed via FastAPI endpoints in src/api/v1/knowledge_filters.py (lines 18‑112). The routes delegate directly to KnowledgeFilterService methods, handling HTTP serialization and authentication. The service is injected through src/dependencies.py, which resolves the clients.opensearch singleton defined in src/config/settings.py.

Summary

• Storage: Knowledge filters are stored as documents in the OpenSearch knowledge_filters index, initialized in src/main.py.
• CRUD Operations: Use create_knowledge_filter, search_knowledge_filters, update_knowledge_filter, and delete_knowledge_filter for document management; all methods refresh the index immediately to ensure consistency.
• Access Control: Each operation validates the requesting user_id against the filter’s owner, allowed_users, and allowed_groups fields.
• Subscriptions: Manage webhook alerts through add_subscription, remove_subscription, and get_filter_subscriptions, which manipulate the subscriptions array inside the filter document.
• Alerting: Integrate with MonitorService.create_knowledge_filter_monitor to register OpenSearch alerts that POST match results to the filter’s webhook URL.
• API Access: Configure filters programmatically via the service class or through the /v1/knowledge-filters/* REST endpoints.

Frequently Asked Questions

How do I initialize the KnowledgeFilterService in my application?

Instantiate the class with a session manager that provides an OpenSearch client connection. According to the OpenRAG source code, the service is typically injected into FastAPI route handlers using the dependency injection system in src/dependencies.py, which resolves the clients.opensearch singleton from src/config/settings.py.

What permissions are required to modify a knowledge filter?

The service enforces ownership and ACL checks on every request. A user can modify or delete a filter only if they are the owner, listed in allowed_users, or belong to a group in allowed_groups. These fields are stored in the OpenSearch document and validated during update_knowledge_filter and delete_knowledge_filter operations.

Can I create multiple webhook subscriptions for a single filter?

Yes. The subscriptions field is an array that can hold multiple subscription objects, each with a unique subscription_id. Use add_subscription to append new webhooks and remove_subscription to delete specific ones without affecting other subscriptions or the filter’s core query configuration.

How does the alerting monitor integrate with knowledge filters?

When you call MonitorService.create_knowledge_filter_monitor, the service converts the filter’s query_data into an OpenSearch Alerting plugin monitor definition. It generates a unique webhook URL (/knowledge-filter/{filter_id}/webhook/{subscription_id}) and registers the monitor to POST match results to that endpoint whenever new documents satisfy the filter query.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}