refactor(moderation): add ModerationService to handle moderation events and forward them via PostResponsesPort

anonflow/moderation/__init__.py

@@ -1,8 +1,11 @@
-from .executor import ModerationExecutor, ModerationPlanner
+from .executor import ModerationExecutor
+from .planner import ModerationPlanner
 from .rule_manager import RuleManager
+from .service import ModerationService
 
 __all__ = [
     "ModerationExecutor",
     "ModerationPlanner",
     "RuleManager",
+    "ModerationService"
 ]

anonflow/moderation/events.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Event:
+    pass
+
+@dataclass(frozen=True)
+class ModerationDecisionEvent(Event):
+    is_approved: bool
+    reason: str
+
+@dataclass(frozen=True)
+class ModerationStartedEvent(Event):
+    pass

anonflow/moderation/executor.py

@@ -1,27 +1,22 @@
 import asyncio
 import logging
 import textwrap
-from typing import AsyncGenerator, Literal, Optional
-
-from anonflow.services.transport.results import (
-    Results,
-    ModerationDecisionResult,
-    ModerationStartedResult
-)
+from typing import AsyncGenerator, Optional, Literal
 
+from .events import Event, ModerationDecisionEvent, ModerationStartedEvent
 from .planner import ModerationPlanner
 
 
 class ModerationExecutor:
-    def __init__(self, planner: ModerationPlanner):
+    def __init__(self, moderation_planner: ModerationPlanner):
         self._logger = logging.getLogger(__name__)
 
-        self.planner = planner
-        self.planner.set_functions(self.moderation_decision)
+        self._moderation_planner = moderation_planner
+        self._moderation_planner.set_functions(self.moderation_decision)
 
     def moderation_decision(self, status: Literal["approve", "reject"], reason: str):
         moderation_map = {"approve": True, "reject": False}
-        return ModerationDecisionResult(is_approved=moderation_map.get(status.lower(), False), reason=reason)
+        return ModerationDecisionEvent(is_approved=moderation_map.get(status.lower(), False), reason=reason)
     moderation_decision.description = textwrap.dedent( # type: ignore
         """
         Processes a message with a moderation decision by status and reason.
@@ -30,15 +25,15 @@ class ModerationExecutor:
         """
     ).strip()
 
-    async def process(self, text: Optional[str] = None, image: Optional[str] = None) -> AsyncGenerator[Results, None]:
-        yield ModerationStartedResult()
+    async def process(self, text: Optional[str] = None, image: Optional[str] = None) -> AsyncGenerator[Event, None]:
+        yield ModerationStartedEvent()
 
-        functions = await self.planner.plan(text, image)
-        function_names = self.planner.get_function_names()
+        functions = await self._moderation_planner.plan(text, image)
+        function_names = self._moderation_planner.get_function_names()
 
-        for func in functions:
-            func_name = func.get("name", "")
-            func_args = func.get("args", {})
+        for function in functions:
+            func_name = function.get("name", "")
+            func_args = function.get("args", {})
 
             method = getattr(self, func_name, None)
 

anonflow/moderation/planner.py

@@ -49,7 +49,7 @@ class ModerationPlanner:
             "http_client": self._client
         }
 
-        self.rule_manager = rule_manager
+        self._rule_manager = rule_manager
 
         self._enabled = False
         self._functions: List[Dict[str, Any]] = []
@@ -111,6 +111,9 @@ class ModerationPlanner:
 
         return "\n".join(lines)
 
+    def _build_rules(self):
+        return "\n\n".join(self._rule_manager.get_rules())
+
     async def _run_omni(self, text: Optional[str] = None, image: Optional[str] = None):
         if self.is_backend_enabled("omni"):
             content = self._build_content(text, image)
@@ -130,6 +133,7 @@ class ModerationPlanner:
 
         if self.is_backend_enabled("gpt"):
             functions_prompt = self._build_functions_prompt(self._functions)
+            rules_prompt = self._build_rules()
 
             output = None
             for attempt in range(self._max_retries + 1):
@@ -153,13 +157,12 @@ class ModerationPlanner:
                                     - Do not omit required arguments.
                                     Available functions:
                                     {functions_prompt}
+
+                                    **RULES:**
+                                    {rules_prompt}
                                     '''
                                 ).strip(),
                             },
-                            {
-                                "role": "system",
-                                "content": "\n\n".join(self.rule_manager.get_rules())
-                            },
                             {
                                 "role": "user",
                                 "content": text
@@ -216,8 +219,8 @@ class ModerationPlanner:
             return
 
         self._functions.clear()
-        for func in functions:
-            sig = inspect.signature(func)
+        for function in functions:
+            sig = inspect.signature(function)
             args = {
                 name: (
                     getattr(param.annotation, "__name__", str(param.annotation))
@@ -227,7 +230,11 @@ class ModerationPlanner:
             }
 
             self._functions.append(
-                {"name": func.__name__, "args": args, "description": func.description or ""}
+                {
+                    "name": function.__name__,
+                    "args": args,
+                    "description": function.description or ""
+                }
             )
 
         function_names = self.get_function_names()

anonflow/moderation/rule_manager.py

@@ -8,16 +8,16 @@ class RuleManager:
     def __init__(self, rules_dir: Path):
         self._logger = logging.getLogger(__name__)
 
-        self.rules_dir = rules_dir
+        self._rules_dir = rules_dir
         self._rules: List[str] = []
 
     def reload(self):
-        if not self.rules_dir.exists():
-            self.rules_dir.mkdir(parents=True, exist_ok=True)
+        if not self._rules_dir.exists():
+            self._rules_dir.mkdir(parents=True, exist_ok=True)
 
         self._rules.clear()
-        for rule_filename in listdir(self.rules_dir):
-            rule_filepath = Path(self.rules_dir / rule_filename).resolve()
+        for rule_filename in listdir(self._rules_dir):
+            rule_filepath = Path(self._rules_dir / rule_filename).resolve()
             with rule_filepath.open(encoding="utf-8") as rule_file:
                 rule = rule_file.read()
                 if rule:
@@ -26,4 +26,4 @@ class RuleManager:
         self._logger.info("Rules loaded. Total=%d", len(self._rules))
 
     def get_rules(self):
-        return self._rules
+        return self._rules.copy()

anonflow/moderation/service.py

@@ -0,0 +1,28 @@
+from typing import Optional
+
+from anonflow.interfaces import PostResponsesPort
+from anonflow.services.transport.types import RequestContext
+
+from .events import ModerationDecisionEvent, ModerationStartedEvent
+from .executor import ModerationExecutor
+
+
+class ModerationService:
+    def __init__(
+        self,
+        responses_port: PostResponsesPort,
+        moderation_executor: ModerationExecutor
+    ):
+        self._responses_port = responses_port
+        self._moderation_executor = moderation_executor
+
+    async def process(self, context: RequestContext, text: Optional[str] = None, image: Optional[str] = None):
+        is_approved = False
+        async for event in self._moderation_executor.process(text, image):
+            if isinstance(event, ModerationDecisionEvent):
+                is_approved = event.is_approved
+                await self._responses_port.post_moderation_decision(context, event.is_approved, event.reason)
+            elif isinstance(event, ModerationStartedEvent):
+                await self._responses_port.post_moderation_started(context)
+
+        return is_approved