feat: introduce post processors

docs/api_reference/agents/index.md

@@ -9,3 +9,5 @@
 ::: ragbits.agents.AgentResultStreaming
 
 ::: ragbits.agents.a2a.server.create_agent_server
+
+::: ragbits.agents.post_processors.base

docs/how-to/agents/use_post_processors.md

@@ -0,0 +1,64 @@
+# How-To: Use Post-Processors with Ragbits Agents
+
+Ragbits Agents can be enhanced with post-processors to intercept, log, filter, and modify their outputs. This guide explains how to implement and use post-processors to customize agent responses.
+
+## Post-Processors Overview
+
+Ragbits provides two types of post-processors:
+
+- **PostProcessor**: Processes the final output after generation, ideal for batch processing.
+- **StreamingPostProcessor**: Processes outputs as they are generated, suitable for real-time applications.
+
+### Implementing a custom Post-Processor
+
+To create a custom post-processor, inherit from the appropriate base class ([`PostProcessor`][ragbits.agents.post_processors.base.PostProcessor] or [`StreamingPostProcessor`][ragbits.agents.post_processors.base.StreamingPostProcessor]) and implement the required method.
+
+#### Post-Processor Example
+
+A non-streaming post-processor applies transformations after the entire content is generated.
+
+```python
+class TruncateProcessor(PostProcessor):
+    def __init__(self, max_length: int = 50) -> None:
+        self.max_length = max_length
+
+    async def process(self, result, agent):
+        content = result.content
+        if len(content) > self.max_length:
+            content = content[:self.max_length] + "... [TRUNCATED]"
+        result.content = content
+        return result
+```
+
+#### Streaming Post-Processor Example
+
+A streaming post-processor can manipulate all information returned during generation, including text, tool calls, etc.
+
+```python
+class UpperCaseStreamingProcessor(StreamingPostProcessor):
+    async def process_streaming(self, chunk, agent):
+        if isinstance(chunk, str):
+            return chunk.upper()
+        return chunk
+```
+
+## Using Post-Processors
+
+To use post-processors, pass them to the `run` or `run_streaming` methods of the `Agent` class. If you pass a non-streaming processor to `run_streaming`, set `allow_non_streaming=True`. This allows streaming processors to handle content piece by piece during generation, while non-streaming processors apply transformations after the entire output is generated.
+
+```python
+async def main() -> None:
+    llm = LiteLLM("gpt-4.1-mini")
+    agent = Agent(llm=llm, prompt="You are a helpful assistant.")
+    post_processors = [
+        UpperCaseStreamingProcessor(),
+        TruncateProcessor(max_length=50),
+    ]
+    stream_result = agent.run_streaming("Tell me about the history of AI.", post_processors=post_processors, allow_non_streaming=True)
+    async for chunk in stream_result:
+        if isinstance(chunk, str):
+            print(chunk, end="")
+    print(f"\nFinal answer:\n{stream_result.content}")
+```
+
+Post-processors offer a flexible way to tailor agent outputs, whether filtering content in real-time or transforming final outputs.

examples/README.md

@@ -20,6 +20,7 @@ All necessary details are provided in the comments at the top of each script.
 | [Multimodal Prompt with PDF Input](/examples/core/prompt/multimodal_with_pdf.py)                 |            [ragbits-core](/packages/ragbits-core)            | Example of how to use the `Prompt` class to answer the question using an LLM with both text and PDF inputs.                                             |
 | [Multimodal Prompt with Few Shots](/examples/core/prompt/multimodal_with_few_shots.py)           |            [ragbits-core](/packages/ragbits-core)            | Example of how to use the `Prompt` class to generate themed text using an LLM with multimodal inputs and few-shot examples.                             |
 | [Tool Use with LLM](/examples/core/llms/tool_use.py)                                             |            [ragbits-core](/packages/ragbits-core)            | Example of how to provide tools and return tool calls from LLM.                                                                                         |
+| [Reasoning with LLM](/examples/core/llms/reasoning.py)                                           |            [ragbits-core](/packages/ragbits-core)            | Example of how to use reasoning with LLM.                                                                                                               |
 | [OpenTelemetry Audit](/examples/core/audit/otel.py)                                              |            [ragbits-core](/packages/ragbits-core)            | Example of how to collect traces and metrics using Ragbits audit module with OpenTelemetry.                                                             |
 | [Logfire Audit](/examples/core/audit/logfire_.py)                                                |            [ragbits-core](/packages/ragbits-core)            | Example of how to collect traces and metrics using Ragbits audit module with Logfire.                                                                   |
 | [Basic Document Search](/examples/document-search/basic.py)                                      | [ragbits-document-search](/packages/ragbits-document-search) | Example of how to use the `DocumentSearch` class to search for documents with the `InMemoryVectorStore` class to store the embeddings.                  |
@@ -38,6 +39,9 @@ All necessary details are provided in the comments at the top of each script.
 | [Recontextualize Last Message](/examples/chat/recontextualize_message.py)                        |            [ragbits-chat](/packages/ragbits-chat)            | Example of how to use the `StandaloneMessageCompressor` compressor to recontextualize the last message in a conversation history.                       |
 | [Agents Tool Use](/examples/agents/tool_use.py)                                                  |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use agent with tools.                                                                                                                 |
 | [Agents OpenAI Native Tool Use](/examples/agents/openai_native_tool_use.py)                      |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use agent with OpenAI native tools.                                                                                                   |
+| [Agents Post Processors](/examples/agents/post_processors.py)                                    |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use post-processors with agent.                                                                                                       |
+| [Agents CLI](/examples/agents/cli_agent.py)                                                      |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use agent in CLI.                                                                                                                     |
 | [MCP Local](/examples/agents/mcp/local.py)                                                       |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use the `Agent` class to connect with a local MCP server.                                                                             |
 | [MCP SSE](/examples/agents/mcp/sse.py)                                                           |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use the `Agent` class to connect with a remote MCP server via SSE.                                                                    |
 | [MCP Streamable HTTP](/examples/agents/mcp/streamable_http.py)                                   |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to use the `Agent` class to connect with a remote MCP server via HTTP.                                                                   |
+| [A2A Orchestration](/examples/agents/a2a/run_orchestrator.py)                                    |          [ragbits-agents](/packages/ragbits-agents)          | Example of how to setup A2A orchestration.                                                                                                              |

examples/agents/post_processors.py

@@ -0,0 +1,97 @@
+"""
+Ragbits Agents Example: Post-Processors
+
+This example demonstrates how to use post-processors with Agent.run() and Agent.run_streaming() methods.
+
+To run the script, execute the following command:
+
+    ```bash
+    uv run examples/agents/post_processors.py
+    ```
+"""
+
+# /// script
+# requires-python = ">=3.10"
+# dependencies = [
+#     "ragbits-core",
+#     "ragbits-agents",
+# ]
+# ///
+
+import asyncio
+from types import SimpleNamespace
+
+from ragbits.agents import Agent, AgentResult, PostProcessor, StreamingPostProcessor, ToolCallResult
+from ragbits.core.llms.base import BasePrompt, ToolCall, Usage
+from ragbits.core.llms.litellm import LiteLLM
+
+
+class CustomStreamingPostProcessor(StreamingPostProcessor):
+    """
+    Streaming post-processor that checks for forbidden words.
+    """
+
+    def __init__(self, forbidden_words: list[str]) -> None:
+        self.forbidden_words = forbidden_words
+
+    async def process_streaming(
+        self, chunk: str | ToolCall | ToolCallResult | SimpleNamespace | BasePrompt | Usage, agent: Agent
+    ) -> str | ToolCall | ToolCallResult | SimpleNamespace | BasePrompt | Usage:
+        """
+        Process chunks during streaming.
+        """
+        if isinstance(chunk, str) and chunk.lower().strip() in self.forbidden_words:
+            return "[FORBIDDEN_WORD]"
+        return chunk
+
+
+class CustomPostProcessor(PostProcessor):
+    """
+    Non-streaming post-processor that truncates the content.
+    """
+
+    def __init__(self, max_length: int = 200) -> None:
+        self.max_length = max_length
+
+    async def process(self, result: AgentResult, agent: Agent) -> AgentResult:
+        """
+        Process the agent result.
+        """
+        content = result.content
+        content_length = len(content)
+
+        if content_length > self.max_length:
+            content = content[: self.max_length]
+            content += f"... [TRUNCATED] ({content_length} > {self.max_length} chars)"
+
+        return AgentResult(
+            content=content,
+            metadata=result.metadata,
+            tool_calls=result.tool_calls,
+            history=result.history,
+            usage=result.usage,
+        )
+
+
+async def main() -> None:
+    """
+    Run the example.
+    """
+    llm = LiteLLM("gpt-4.1-mini")
+    agent: Agent = Agent(llm=llm, prompt="You are a helpful assistant.")
+    stream_result = agent.run_streaming(
+        "What is Python?",
+        post_processors=[
+            CustomStreamingPostProcessor(forbidden_words=["python"]),
+            CustomPostProcessor(max_length=200),
+        ],
+        allow_non_streaming=True,
+    )
+    async for chunk in stream_result:
+        if isinstance(chunk, str):
+            print(chunk, end="")
+    print(f"\n\nFinal answer:\n{stream_result.content}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

mkdocs.yml

@@ -40,6 +40,7 @@ nav:
         - "Define and use agents": how-to/agents/define_and_use_agents.md
         - "Provide tools with MCP": how-to/agents/provide_mcp_tools.md
         - "Serve agents with A2A": how-to/agents/serve_ragbits_agents.md
+        - "Use post-processors": how-to/agents/use_post_processors.md
       - Guardrails:
         - "Setup guardrails": how-to/guardrails/use_guardrails.md
       - Chatbots:

packages/ragbits-agents/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Support wrapping downstream agents as tools (#818)
 - Add syntax sugar allowing easier Agents definition (#820)
+- Add post-processors (#821)
 
 ## 1.3.0 (2025-09-11)
 ### Changed

packages/ragbits-agents/src/ragbits/agents/__init__.py

@@ -7,6 +7,7 @@
     AgentRunContext,
     ToolCallResult,
 )
+from ragbits.agents.post_processors.base import PostProcessor, StreamingPostProcessor
 from ragbits.agents.types import QuestionAnswerAgent, QuestionAnswerPromptInput, QuestionAnswerPromptOutput
 
 __all__ = [
@@ -16,8 +17,10 @@
     "AgentResult",
     "AgentResultStreaming",
     "AgentRunContext",
+    "PostProcessor",
     "QuestionAnswerAgent",
     "QuestionAnswerPromptInput",
     "QuestionAnswerPromptOutput",
+    "StreamingPostProcessor",
     "ToolCallResult",
 ]