Skip to content

Postprocessor Manager

📘 Postprocessor and PostprocessorManager Documentation

The Postprocessor and PostprocessorManager classes are responsible for refining and modifying responses generated by the orchestrator. Postprocessors are modular components that apply specific transformations or enhancements to the generated text. The PostprocessorManager orchestrates the execution of these postprocessors, ensuring they are applied.

🔑 Postprocessor Overview

The Postprocessor is an abstract class that defines the structure and behavior of individual postprocessors. Each postprocessor must implement the following methods to be compatible with the PostprocessorManager:

  1. run: Executes the postprocessing logic on the response.
  2. apply_postprocessing_to_response: Applies the postprocessing changes to the response if necessary.
  3. remove_from_text: Removes any postprocessing artifacts from the text, ensuring clean input for subsequent LLM rounds.

Postprocessor Class Definition:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class Postprocessor(ABC):
    def __init__(self, name: str):
        self.name = name

    def get_name(self) -> str:
        """Returns the name of the postprocessor."""
        return self.name

    async def run(self, loop_response: LanguageModelStreamResponse) -> str:
        """Executes the postprocessing logic."""
        raise NotImplementedError("Subclasses must implement this method.")

    async def apply_postprocessing_to_response(
        self, loop_response: LanguageModelStreamResponse
    ) -> bool:
        """Applies postprocessing changes to the response."""
        raise NotImplementedError(
            "Subclasses must implement this method to apply post-processing to the response."
        )

    async def remove_from_text(self, text) -> str:
        """Removes postprocessing artifacts from the text."""
        raise NotImplementedError(
            "Subclasses must implement this method to remove post-processing from the message."
        )

🔑 PostprocessorManager Overview

The PostprocessorManager is responsible for managing and executing postprocessors. It ensures that all configured postprocessors are run in parallel and their results are applied sequentially to the response. Additionally, it provides functionality to remove postprocessing artifacts from text when necessary.

🛠️ Key Functionalities of PostprocessorManager

1. Postprocessor Management

  • add_postprocessor(postprocessor: Postprocessor)
    Adds a postprocessor to the manager's collection.

    1
2
    def add_postprocessor(self, postprocessor: Postprocessor):
    self._postprocessors.append(postprocessor)

  • get_postprocessors(name: str) -> list[Postprocessor]
    Retrieves the list of all registered postprocessors.

    1
2
    def get_postprocessors(self, name: str) -> list[Postprocessor]:
    return self._postprocessors

2. Postprocessor Execution

  • run_postprocessors(loop_response: LanguageModelStreamResponse)
    Executes all registered postprocessors asynchronously. Results are applied sequentially to the response if modifications are detected.
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
    async def run_postprocessors(
    self,
    loop_response: LanguageModelStreamResponse,
) -> None:
    # Executes postprocessors in parallel
    tasks = [
        task_executor.execute_async(
            self.execute_postprocessors,
            loop_response=loop_response,
            postprocessor_instance=postprocessor,
        )
        for postprocessor in self._postprocessors
    ]
    postprocessor_results = await asyncio.gather(*tasks)

    # Applies modifications sequentially
    modification_results = [
        postprocessor.apply_postprocessing_to_response(loop_response)
        for postprocessor in self._postprocessors
    ]
    has_been_modified = any(modification_results)

    if has_been_modified:
        self._chat_service.modify_assistant_message(
            content=loop_response.message.text,
            message_id=loop_response.message.id,
        )

async def execute_postprocessors(
    self,
    loop_response: LanguageModelStreamResponse,
    postprocessor_instance: Postprocessor,
) -> None:
    await postprocessor_instance.run(loop_response)

3. Text Cleanup

  • remove_from_text(text: str) -> str
    Iterates through all registered postprocessors and removes their artifacts from the text. This ensures that postprocessing-generated content does not interfere with subsequent LLM rounds.
    1
2
3
4
5
6
7
    async def remove_from_text(
    self,
    text: str,
) -> str:
    for postprocessor in self._postprocessors:
        text = await postprocessor.remove_from_text(text)
    return text

🛠️ Summary of Responsibilities

  1. Postprocessor:
  2. Defines the structure and behavior of individual postprocessors.

  3. Implements methods for running, applying, and removing postprocessing logic.

  4. PostprocessorManager:

  5. Manages a collection of postprocessors.
  6. Executes postprocessors in parallel and applies their results sequentially.
  7. Provides functionality to clean up postprocessing artifacts from text.

This documentation provides a concise and focused explanation of the Postprocessor and PostprocessorManager classes, ensuring developers can implement and manage postprocessing logic effectively. Let me know if further refinements are needed!