Move topic conditions to node publisher routes

grafi/agents/react_agent.py

@@ -14,13 +14,13 @@
 from grafi.common.models.async_result import async_func_wrapper
 from grafi.common.models.invoke_context import InvokeContext
 from grafi.common.models.message import Message
+from grafi.nodes.conditions import has_text_response
+from grafi.nodes.conditions import has_tool_call
 from grafi.nodes.node import Node
 from grafi.runtime import GrafiRuntime
 from grafi.tools.function_calls.function_call_tool import FunctionCallTool
 from grafi.tools.function_calls.impl.tavily_tool import TavilyTool
 from grafi.tools.llms.impl.openai_tool import OpenAITool
-from grafi.topics.conditions import has_text_response
-from grafi.topics.conditions import has_tool_call
 from grafi.topics.expressions.subscription_builder import SubscriptionBuilder
 from grafi.topics.topic_impl.input_topic import InputTopic
 from grafi.topics.topic_impl.output_topic import OutputTopic
@@ -64,18 +64,12 @@ def builder(cls) -> "ReActAgentBuilder":
         return ReActAgentBuilder(cls)
 
     def _construct_workflow(self) -> "ReActAgent":
-        function_call_topic = Topic(
-            name="function_call_topic",
-            condition=has_tool_call,
-        )
+        function_call_topic = Topic(name="function_call_topic")
         function_result_topic = Topic(name="function_result_topic")
 
         agent_input_topic = InputTopic(name="agent_input_topic")
 
-        agent_output_topic = OutputTopic(
-            name="agent_output_topic",
-            condition=has_text_response,
-        )
+        agent_output_topic = OutputTopic(name="agent_output_topic")
 
         llm_node = (
             Node.builder()
@@ -96,21 +90,21 @@ def _construct_workflow(self) -> "ReActAgent":
                 .system_message(self.system_prompt)
                 .build()
             )
-            .publish_to(function_call_topic)
-            .publish_to(agent_output_topic)
+            .publish_to(function_call_topic, when=has_tool_call)
+            .publish_to(agent_output_topic, when=has_text_response)
             .build()
         )
 
         # Create a function call node
 
-        function_call_node = Node(
-            name="Node",
-            type="Node",
-            tool=self.function_call_tool,
-            subscribed_expressions=[
-                SubscriptionBuilder().subscribed_to(function_call_topic).build()
-            ],
-            publish_to=[function_result_topic],
+        function_call_node = (
+            Node.builder()
+            .name("Node")
+            .type("Node")
+            .tool(self.function_call_tool)
+            .subscribe(SubscriptionBuilder().subscribed_to(function_call_topic).build())
+            .publish_to(function_result_topic)
+            .build()
         )
 
         # Create a workflow and add the nodes

grafi/topics/conditions.py → grafi/nodes/conditions.py

@@ -1,8 +1,9 @@
-"""Reusable topic-condition predicates.
+"""Reusable routing predicates.
 
-A topic ``condition`` decides whether an event is published to that topic. These
-named, importable predicates cover the common routing checks so workflows can
-reference them (``condition=has_tool_call``) instead of embedding inline lambdas.
+A publisher route's ``when`` decides whether an outgoing event is delivered to
+that route's topic. These named, importable predicates cover the common content
+checks so workflows can reference them (``when=has_tool_call``) instead of
+embedding inline lambdas.
 
 Named predicates are preferred over lambdas because they:
 
@@ -18,6 +19,15 @@
 from grafi.common.events.topic_events.publish_to_topic_event import PublishToTopicEvent
 
 
+def always_true(_: PublishToTopicEvent) -> bool:
+    """Default routing predicate: take the route for every event.
+
+    A named module-level function (rather than an inline ``lambda``) so it
+    serializes as a tiny import reference instead of inline code.
+    """
+    return True
+
+
 def has_tool_call(event: PublishToTopicEvent) -> bool:
     """Publish only when the last message requests a tool/function call."""
     return event.data[-1].tool_calls is not None

grafi/nodes/node.py

@@ -12,6 +12,7 @@
 from grafi.common.models.invoke_context import InvokeContext
 from grafi.nodes.node_base import NodeBase
 from grafi.nodes.node_base import NodeBaseBuilder
+from grafi.nodes.node_base import PublisherRouter
 from grafi.tools.command import Command
 from grafi.tools.tool_factory import ToolFactory
 from grafi.topics.expressions.topic_expression import CombinedExpr
@@ -73,17 +74,21 @@ def to_dict(self) -> dict[str, Any]:
         }
 
     @classmethod
-    async def from_dict(
+    async def _builder_from_dict(
         cls, node_dict: Dict[str, Any], topics: Dict[str, TopicBase]
-    ) -> "Node":
-        """Create a Node instance from a dictionary representation."""
+    ) -> NodeBaseBuilder:
+        """Reconstruct this node's builder from a dictionary representation.
+
+        Subclasses extend deserialization by calling this and then applying
+        their own fields to the returned builder.
+        """
         from openinference.semconv.trace import OpenInferenceSpanKindValues
 
         # Deserialize the tool
         tool = await ToolFactory.from_dict(node_dict["tool"])
 
         node_builder = (
-            Node.builder()
+            cls.builder()
             .name(node_dict["name"])
             .type(node_dict["type"])
             .oi_span_type(
@@ -103,11 +108,22 @@ async def from_dict(
             elif "op" in expr_dict:
                 node_builder.subscribe(await CombinedExpr.from_dict(expr_dict, topics))
 
-        for topic_name in node_dict["publish_to"]:
-            # Check if topic already exists in topics dict
-            if topic_name in topics:
-                node_builder.publish_to(topics[topic_name])
-            else:
-                raise ValueError(f"Unknown topic: {topic_name}")
+        if "publisher_router" not in node_dict:
+            raise ValueError(
+                "Node manifest is missing required key 'publisher_router'. It "
+                "replaced the legacy 'publish_to' topic-name list; regenerate "
+                "the manifest from the current assistant."
+            )
+        node_builder.publisher_router(
+            PublisherRouter.from_dict(node_dict["publisher_router"], topics)
+        )
+
+        return node_builder
 
+    @classmethod
+    async def from_dict(
+        cls, node_dict: Dict[str, Any], topics: Dict[str, TopicBase]
+    ) -> "Node":
+        """Create a Node instance from a dictionary representation."""
+        node_builder = await cls._builder_from_dict(node_dict, topics)
         return node_builder.build()

grafi/nodes/node_base.py

@@ -1,18 +1,22 @@
 from typing import Any
 from typing import AsyncGenerator
+from typing import Callable
 from typing import Dict
 from typing import List
 from typing import Optional
 from typing import Self
 from typing import TypeVar
 from typing import Union
 
+from loguru import logger
 from openinference.semconv.trace import OpenInferenceSpanKindValues
 from pydantic import BaseModel
 from pydantic import ConfigDict
 from pydantic import Field
 from pydantic import PrivateAttr
 
+from grafi.common.callable_ref import deserialize_callable
+from grafi.common.callable_ref import serialize_callable
 from grafi.common.events.topic_events.consume_from_topic_event import (
     ConsumeFromTopicEvent,
 )
@@ -21,6 +25,7 @@
 from grafi.common.models.base_builder import BaseBuilder
 from grafi.common.models.default_id import default_id
 from grafi.common.models.invoke_context import InvokeContext
+from grafi.nodes.conditions import always_true
 from grafi.tools.command import Command
 from grafi.tools.tool import Tool
 from grafi.topics.expressions.topic_expression import SubExpr
@@ -29,6 +34,97 @@
 from grafi.topics.topic_base import TopicBase
 
 
+class PublishRoute(BaseModel):
+    """One destination in the publisher router's table.
+
+    ``when`` is the CONTENT predicate (what used to live on ``topic.condition``).
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    topic: TopicBase  # destination (definition-time instance)
+    when: Callable[[PublishToTopicEvent], bool] = always_true  # content predicate
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "topic": self.topic.name,
+            "when": serialize_callable(self.when),
+        }
+
+    @classmethod
+    def from_dict(
+        cls, data: Dict[str, Any], topics: Dict[str, TopicBase]
+    ) -> "PublishRoute":
+        topic_name = data["topic"]
+        if topic_name not in topics:
+            raise ValueError(
+                f"Publisher route references unknown topic {topic_name!r}. "
+                f"Available topics: {sorted(topics)}."
+            )
+        raw = data.get("when")
+        when = (
+            deserialize_callable(raw, context=f"route '{topic_name}' when")
+            if raw
+            else always_true
+        )
+        return cls(
+            topic=topics[topic_name],
+            when=when,
+        )
+
+
+class PublisherRouter(BaseModel):
+    """The node's output routing table; ``publish_to`` derives from its routes."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    routes: List[PublishRoute] = Field(default_factory=list)
+
+    def select(self, event: PublishToTopicEvent) -> List[str]:
+        """Destination topic names for an outgoing event, routed by content."""
+        return [r.topic.name for r in self.routes if self._condition_ok(r.when, event)]
+
+    @staticmethod
+    def _condition_ok(
+        when: Callable[[PublishToTopicEvent], bool], event: PublishToTopicEvent
+    ) -> bool:
+        """Evaluate a route's content predicate with the topic error policy.
+
+        Mirrors the old ``TopicBase.publish_data``: an ``IndexError``/``KeyError``
+        is the expected "not met" signal for predicates that index possibly-empty
+        data; any other exception is a probable bug, surfaced at WARNING but
+        still treated as not-met so one faulty predicate cannot halt the
+        workflow.
+        """
+        try:
+            return bool(when(event))
+        except (IndexError, KeyError) as e:
+            logger.debug(f"Route predicate not met ({type(e).__name__}: {e}).")
+            return False
+        except Exception as e:  # noqa: BLE001 - defensive: a user predicate bug
+            logger.warning(
+                f"Route predicate raised an unexpected {type(e).__name__}: {e}. "
+                "Treating as not published; check the publisher route's `when`."
+            )
+            return False
+
+    @property
+    def publish_to(self) -> List[TopicBase]:
+        """Every destination topic (the node's single output declaration)."""
+        return [r.topic for r in self.routes]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"routes": [r.to_dict() for r in self.routes]}
+
+    @classmethod
+    def from_dict(
+        cls, data: Dict[str, Any], topics: Dict[str, TopicBase]
+    ) -> "PublisherRouter":
+        return cls(
+            routes=[PublishRoute.from_dict(r, topics) for r in data.get("routes", [])]
+        )
+
+
 class NodeBase(BaseModel):
     """Abstract base class for nodes in a graph-based agent system."""
 
@@ -40,11 +136,16 @@ class NodeBase(BaseModel):
     tool: Optional[Tool] = Field(default=None)
     oi_span_type: OpenInferenceSpanKindValues = OpenInferenceSpanKindValues.CHAIN
     subscribed_expressions: List[SubExpr] = Field(default_factory=list)
-    publish_to: List[TopicBase] = Field(default_factory=list)
+    publisher_router: PublisherRouter = Field(default_factory=PublisherRouter)
 
     _subscribed_topics: Dict[str, TopicBase] = PrivateAttr(default_factory=dict)
     _command: Optional[Command] = PrivateAttr(default=None)
 
+    @property
+    def publish_to(self) -> List[TopicBase]:
+        """Every destination topic (derived from the publisher router's routes)."""
+        return self.publisher_router.publish_to
+
     @property
     def command(self) -> Optional[Command]:
         """Access the internal command (for backward compatibility).
@@ -128,7 +229,7 @@ def to_dict(self) -> dict[str, Any]:
             "subscribed_expressions": [
                 expr.to_dict() for expr in self.subscribed_expressions
             ],
-            "publish_to": [topic.name for topic in self.publish_to],
+            "publisher_router": self.publisher_router.to_dict(),
             "command": self.command.to_dict() if self.command else None,
         }
 
@@ -246,16 +347,37 @@ def subscribe(self, subscribe_to: Union[TopicBase, SubExpr]) -> Self:
             )
         return self
 
-    def publish_to(self, topic: TopicBase) -> Self:
-        """Add a topic that this node will publish results to.
+    def publish_to(
+        self,
+        topic: TopicBase,
+        when: Optional[Callable[[PublishToTopicEvent], bool]] = None,
+    ) -> Self:
+        """Add a publisher route to a topic this node publishes to.
 
         Args:
             topic: The topic to publish output events to.
+            when: Optional content predicate (defaults to "always"); the route is
+                taken when it returns True for the outgoing event.
+
+        Returns:
+            Self for method chaining.
+        """
+        if "publisher_router" not in self.kwargs:
+            self.kwargs["publisher_router"] = PublisherRouter()
+        route_kwargs: Dict[str, Any] = {"topic": topic}
+        if when is not None:
+            route_kwargs["when"] = when
+        self.kwargs["publisher_router"].routes.append(PublishRoute(**route_kwargs))
+        return self
+
+    def publisher_router(self, publisher_router: PublisherRouter) -> Self:
+        """Set the whole publisher router (used when deserializing a node).
+
+        Args:
+            publisher_router: The reconstructed output routing table.
 
         Returns:
             Self for method chaining.
         """
-        if "publish_to" not in self.kwargs:
-            self.kwargs["publish_to"] = []
-        self.kwargs["publish_to"].append(topic)
+        self.kwargs["publisher_router"] = publisher_router
         return self