Merge branch 'main' into fix-revert-so-easier-for-big-pr

Author: sicoyle

dapr_agents/document/__init__.py

@@ -1,7 +1,11 @@
+from typing import TYPE_CHECKING
+
 from .embedder import NVIDIAEmbedder, OpenAIEmbedder, SentenceTransformerEmbedder
 from .fetcher import ArxivFetcher
 from .reader import PyMuPDFReader, PyPDFReader
-from .splitter import TextSplitter
+
+if TYPE_CHECKING:
+    from .splitter import TextSplitter
 
 __all__ = [
     "ArxivFetcher",
@@ -12,3 +16,12 @@
     "SentenceTransformerEmbedder",
     "NVIDIAEmbedder",
 ]
+
+
+def __getattr__(name: str):
+    """Lazy import for optional dependencies."""
+    if name == "TextSplitter":
+        from .splitter import TextSplitter
+
+        return TextSplitter
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

dapr_agents/llm/utils/request.py

@@ -1,12 +1,13 @@
-from typing import Dict, Any, Optional, List, Type, Union, Iterable, Literal
-from dapr_agents.prompt.prompty import Prompty, PromptyHelper
-from dapr_agents.types.message import BaseMessage
-from dapr_agents.llm.utils.structure import StructureHandler
-from dapr_agents.tool.utils.tool import ToolHelper
+import logging
+from typing import Any, Dict, Iterable, List, Literal, Optional, Type, Union
+
 from pydantic import BaseModel, ValidationError
-from dapr_agents.tool.base import AgentTool
 
-import logging
+from dapr_agents.llm.utils.structure import StructureHandler
+from dapr_agents.prompt.prompty import Prompty, PromptyHelper
+from dapr_agents.tool.base import AgentTool
+from dapr_agents.tool.utils.tool import ToolHelper
+from dapr_agents.types.message import BaseMessage
 
 logger = logging.getLogger(__name__)
 
@@ -100,46 +101,38 @@ def process_params(
         Prepare request parameters for the language model.
 
         Args:
-            params: Parameters for the request.
-            llm_provider: The LLM provider to use (e.g., 'openai').
-            tools: List of tools to include in the request.
-            response_format: Either a Pydantic model (for function calling)
-                            or a JSON Schema definition/dict (for raw JSON structured output).
-            structured_mode: The mode of structured output: 'json' or 'function_call'.
-                            Defaults to 'json'.
+            params: Raw request params (messages/inputs, model, etc.).
+            llm_provider: Provider key, e.g. "openai", "dapr".
+            tools: Tools to expose to the model (AgentTool or already-shaped dicts).
+            response_format:
+                - If structured_mode == "json": a JSON Schema dict or a Pydantic model
+                (we'll convert) to request raw JSON output.
+                - If structured_mode == "function_call": a Pydantic model describing
+                the function/tool signature for model-side function calling.
+            structured_mode: "json" for raw JSON structured output,
+                            "function_call" for tool/function calling.
 
         Returns:
-            Dict[str, Any]: Prepared request parameters.
+            A params dict ready for the target provider.
         """
+
+        # Tools
         if tools:
             logger.info("Tools are available in the request.")
-            # Convert AgentTool objects to dict format for the provider
-            tool_dicts = []
-            for tool in tools:
-                if isinstance(tool, AgentTool):
-                    tool_dicts.append(
-                        ToolHelper.format_tool(tool, tool_format=llm_provider)
-                    )
-                else:
-                    tool_dicts.append(
-                        ToolHelper.format_tool(tool, tool_format=llm_provider)
-                    )
-            params["tools"] = tool_dicts
+            params["tools"] = [
+                ToolHelper.format_tool(t, tool_format=llm_provider) for t in tools
+            ]
 
+        # Structured output
         if response_format:
-            logger.info(f"Structured Mode Activated! Mode={structured_mode}.")
-            # Add system message for JSON formatting
-            # This is necessary for the response formatting of the data to work correctly when a user has a function call response format.
-            inputs = params.get("inputs", [])
-            inputs.insert(
-                0,
-                {
-                    "role": "system",
-                    "content": "You must format your response as a valid JSON object matching the provided schema. Do not include any explanatory text or markdown formatting.",
-                },
-            )
-            params["inputs"] = inputs
+            logger.info(f"Structured Mode Activated! mode={structured_mode}")
+
+            # If we're on Dapr, we cannot rely on OpenAI-style `response_format`.
+            # Add a small system nudge to enforce JSON-only output so we can parse reliably.
+            if llm_provider == "dapr":
+                params = StructureHandler.ensure_json_only_system_prompt(params)
 
+            # Generate provider-specific request params
             params = StructureHandler.generate_request(
                 response_format=response_format,
                 llm_provider=llm_provider,

dapr_agents/llm/utils/structure.py

@@ -627,3 +627,30 @@ def validate_against_signature(result: Any, expected_type: Any) -> Any:
             return adapter.validate_python(result)
         except ValidationError as e:
             raise TypeError(f"Validation failed for type {expected_type}: {e}")
+
+    @staticmethod
+    def ensure_json_only_system_prompt(params: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Dapr's chat client (today) does NOT forward OpenAI-style `response_format`
+        (e.g., {"type":"json_schema", ...}). That means the model won't be hard-constrained
+        to your schema. As a fallback, we prepend a system message that instructs the
+        model to return strict JSON so downstream parsing doesn't break.
+
+        Note:
+        - Dapr uses "inputs" (not "messages") for the message array.
+        - If "inputs" isn't present (future providers), we fall back to "messages".
+        """
+        collection_key = "inputs" if "inputs" in params else "messages"
+        msgs = list(params.get(collection_key, []))
+        msgs.insert(
+            0,
+            {
+                "role": "system",
+                "content": (
+                    "Return ONLY a valid JSON object that matches the provided schema. "
+                    "No markdown, no code fences, no explanations—JSON object only."
+                ),
+            },
+        )
+        params[collection_key] = msgs
+        return params

dapr_agents/workflow/base.py

@@ -7,7 +7,7 @@
 import sys
 import uuid
 from datetime import datetime, timezone
-from typing import Any, Callable, Dict, List, Optional, TypeVar, Union
+from typing import Any, Callable, Dict, List, Optional, TypeVar, Union, Sequence
 
 from dapr.ext.workflow import (
     DaprWorkflowClient,
@@ -16,7 +16,7 @@
 )
 from dapr.ext.workflow.workflow_state import WorkflowState
 from durabletask import task as dtask
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from dapr_agents.agents.base import ChatClientBase
 from dapr_agents.llm.utils.defaults import get_default_llm
@@ -46,6 +46,14 @@ class WorkflowApp(BaseModel, SignalHandlingMixin):
         default=300,
         description="Default timeout duration in seconds for workflow tasks.",
     )
+    grpc_max_send_message_length: Optional[int] = Field(
+        default=None,
+        description="Maximum message length in bytes for gRPC send operations. Default is 4MB if not specified. Useful for AI workflows with large payloads (e.g., images).",
+    )
+    grpc_max_receive_message_length: Optional[int] = Field(
+        default=None,
+        description="Maximum message length in bytes for gRPC receive operations. Default is 4MB if not specified. Useful for AI workflows with large payloads (e.g., images).",
+    )
 
     # Initialized in model_post_init
     wf_runtime: Optional[WorkflowRuntime] = Field(
@@ -68,10 +76,30 @@ class WorkflowApp(BaseModel, SignalHandlingMixin):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
+    @model_validator(mode="before")
+    def validate_grpc_chanell_options(cls, values: Any):
+        if not isinstance(values, dict):
+            return values
+
+        if values.get("grpc_max_send_message_length") is not None:
+            if values["grpc_max_send_message_length"] < 0:
+                raise ValueError("grpc_max_send_message_length must be greater than 0")
+
+        if values.get("grpc_max_receive_message_length") is not None:
+            if values["grpc_max_receive_message_length"] < 0:
+                raise ValueError(
+                    "grpc_max_receive_message_length must be greater than 0"
+                )
+
+        return values
+
     def model_post_init(self, __context: Any) -> None:
         """
         Initialize the Dapr workflow runtime and register tasks & workflows.
         """
+        if self.grpc_max_send_message_length or self.grpc_max_receive_message_length:
+            self._configure_grpc_channel_options()
+
         # Initialize LLM first
         if self.llm is None:
             self.llm = get_default_llm()
@@ -92,6 +120,95 @@ def model_post_init(self, __context: Any) -> None:
 
         super().model_post_init(__context)
 
+    def _configure_grpc_channel_options(self) -> None:
+        """
+        Configure gRPC channel options before workflow runtime initialization.
+        This patches the durabletask internal channel factory to support custom message size limits.
+
+        This is particularly useful for AI-powered workflows that may need to handle large payloads
+        such as images, which can exceed the default 4MB gRPC message size limit.
+        """
+        try:
+            import grpc
+            from durabletask.internal import shared
+
+            # Create custom options list
+            options = []
+            if self.grpc_max_send_message_length:
+                options.append(
+                    ("grpc.max_send_message_length", self.grpc_max_send_message_length)
+                )
+                logger.debug(
+                    f"Configured gRPC max_send_message_length: {self.grpc_max_send_message_length} bytes ({self.grpc_max_send_message_length / (1024 * 1024):.2f} MB)"
+                )
+            if self.grpc_max_receive_message_length:
+                options.append(
+                    (
+                        "grpc.max_receive_message_length",
+                        self.grpc_max_receive_message_length,
+                    )
+                )
+                logger.debug(
+                    f"Configured gRPC max_receive_message_length: {self.grpc_max_receive_message_length} bytes ({self.grpc_max_receive_message_length / (1024 * 1024):.2f} MB)"
+                )
+
+            # Patch the function to include our custom options
+            def get_grpc_channel_with_options(
+                host_address: Optional[str],
+                secure_channel: bool = False,
+                interceptors: Optional[Sequence["grpc.ClientInterceptor"]] = None,
+            ):
+                # This is a copy of the original get_grpc_channel function in durabletask.internal.shared at
+                # https://github.com/dapr/durabletask-python/blob/7070cb07d07978d079f8c099743ee4a66ae70e05/durabletask/internal/shared.py#L30C1-L61C19
+                # but with my option overrides applied above.
+                if host_address is None:
+                    host_address = shared.get_default_host_address()
+
+                for protocol in getattr(shared, "SECURE_PROTOCOLS", []):
+                    if host_address.lower().startswith(protocol):
+                        secure_channel = True
+                        # remove the protocol from the host name
+                        host_address = host_address[len(protocol) :]
+                        break
+
+                for protocol in getattr(shared, "INSECURE_PROTOCOLS", []):
+                    if host_address.lower().startswith(protocol):
+                        secure_channel = False
+                        # remove the protocol from the host name
+                        host_address = host_address[len(protocol) :]
+                        break
+
+                # Create the base channel
+                if secure_channel:
+                    credentials = grpc.ssl_channel_credentials()
+                    channel = grpc.secure_channel(
+                        host_address, credentials, options=options
+                    )
+                else:
+                    channel = grpc.insecure_channel(host_address, options=options)
+
+                # Apply interceptors ONLY if they exist
+                if interceptors:
+                    channel = grpc.intercept_channel(channel, *interceptors)
+
+                return channel
+
+            # Replace the function
+            shared.get_grpc_channel = get_grpc_channel_with_options
+
+            logger.debug(
+                "Successfully patched durabletask gRPC channel factory with custom options"
+            )
+
+        except ImportError as e:
+            logger.error(
+                f"Failed to import required modules for gRPC configuration: {e}"
+            )
+            raise
+        except Exception as e:
+            logger.error(f"Failed to configure gRPC channel options: {e}")
+            raise
+
     def graceful_shutdown(self) -> None:
         """
         Perform graceful shutdown operations for the WorkflowApp.

dapr_agents/workflow/decorators/__init__.py

@@ -1,5 +1,13 @@
 from .core import task, workflow
 from .fastapi import route
 from .messaging import message_router
+from .activities import llm_activity, agent_activity
 
-__all__ = ["workflow", "task", "route", "message_router"]
+__all__ = [
+    "workflow",
+    "task",
+    "route",
+    "message_router",
+    "llm_activity",
+    "agent_activity",
+]