2424from .schemas import (
2525 AgentTaskResponse ,
2626 BroadcastMessage ,
27+ InternalTriggerAction ,
2728 TriggerAction ,
2829)
2930from .state import (
@@ -321,6 +322,30 @@ def tool_calling_workflow(self, ctx: DaprWorkflowContext, message: TriggerAction
321322 # Return the final response message
322323 return final_msg
323324
325+ @message_router
326+ @workflow (name = "AgenticWorkflow" )
327+ def internal_trigger_workflow (
328+ self , ctx : DaprWorkflowContext , message : InternalTriggerAction
329+ ):
330+ """
331+ Handles InternalTriggerAction messages by treating them the same as TriggerAction.
332+ This prevents self-triggering loops while allowing orchestrators to trigger agents.
333+
334+ Args:
335+ ctx (DaprWorkflowContext): The workflow context for the current execution.
336+ message (InternalTriggerAction): The internal trigger message from an orchestrator.
337+
338+ Returns:
339+ Dict[str, Any]: The final response message when the workflow completes.
340+ """
341+ # Convert InternalTriggerAction to TriggerAction format and delegate to the main workflow
342+ trigger_message = TriggerAction (
343+ task = message .task ,
344+ workflow_instance_id = message .workflow_instance_id ,
345+ source = "orchestrator" , # Default source for internal triggers
346+ )
347+ return self .tool_calling_workflow (ctx , trigger_message )
348+
324349 def get_source_or_default (self , source : str ):
325350 # Set default source if not provided (for direct run() calls)
326351 if not source :
@@ -771,14 +796,14 @@ def finalize_workflow(
771796 @message_router (broadcast = True )
772797 async def process_broadcast_message (self , message : BroadcastMessage ):
773798 """
774- Processes a broadcast message, filtering out messages sent by the same agent
775- and updating local memory with valid messages .
799+ Processes a broadcast message by filtering out messages from the same agent,
800+ storing valid messages in memory, and triggering the agent's workflow if needed .
776801
777802 Args:
778803 message (BroadcastMessage): The received broadcast message.
779804
780805 Returns:
781- None: The function updates the agent's memory and ignores unwanted messages .
806+ None: The function updates the agent's memory and triggers a workflow .
782807 """
783808 try :
784809 # Extract metadata safely from message["_message_metadata"]
@@ -819,9 +844,29 @@ async def process_broadcast_message(self, message: BroadcastMessage):
819844 # Save the state after processing the broadcast message
820845 self .save_state ()
821846
847+ # Trigger agent workflow to respond to the broadcast message
848+ workflow_instance_id = metadata .get ("workflow_instance_id" )
849+ if workflow_instance_id :
850+ # Create a TriggerAction to start the agent's workflow
851+ trigger_message = TriggerAction (
852+ task = message .content , workflow_instance_id = workflow_instance_id
853+ )
854+ trigger_message ._message_metadata = {
855+ "source" : metadata .get ("source" , "unknown" ),
856+ "type" : "BroadcastMessage" ,
857+ "workflow_instance_id" : workflow_instance_id ,
858+ }
859+
860+ # Start the agent's workflow
861+ await self .run_and_monitor_workflow_async (
862+ workflow = "ToolCallingWorkflow" , input = trigger_message
863+ )
864+
822865 except Exception as e :
823866 logger .error (f"Error processing broadcast message: { e } , exc_info = True )
824867
868+ # TODO: we need to better design context history management. Context engineering is important,
869+ # and too much context can derail the agent.
825870 def _construct_messages_with_instance_history (
826871 self , instance_id : str , input_data : Union [str , Dict [str , Any ]]
827872 ) -> List [Dict [str , Any ]]:
@@ -843,14 +888,52 @@ def _construct_messages_with_instance_history(
843888 )
844889
845890 # Get instance-specific chat history instead of global memory
891+ if self .state is None :
892+ logger .warning (
893+ f"Agent state is None for instance { instance_id }
894+ )
895+ self .state = {}
896+
846897 instance_data = self .state .get ("instances" , {}).get (instance_id )
847898 if instance_data is not None :
848899 instance_messages = instance_data .get ("messages" , [])
849900 else :
850901 instance_messages = []
851902
852- # Convert instance messages to the format expected by prompt template
903+ # Always include long-term memory (chat_history) for context
904+ # This ensures agents have access to broadcast messages and persistent context
905+ long_term_memory_data = self .state .get ("chat_history" , [])
906+
907+ # Convert long-term memory to dict format for LLM consumption
908+ long_term_memory_messages = []
909+ for msg in long_term_memory_data :
910+ if isinstance (msg , dict ):
911+ long_term_memory_messages .append (msg )
912+ elif hasattr (msg , "model_dump" ):
913+ long_term_memory_messages .append (msg .model_dump ())
914+
915+ # For broadcast-triggered workflows, also include additional context memory
916+ source = instance_data .get ("source" ) if instance_data else None
917+ additional_context_messages = []
918+ if source and source != "direct" :
919+ # Include additional context memory for broadcast-triggered workflows
920+ context_memory_data = self .memory .get_messages ()
921+ for msg in context_memory_data :
922+ if isinstance (msg , dict ):
923+ additional_context_messages .append (msg )
924+ elif hasattr (msg , "model_dump" ):
925+ additional_context_messages .append (msg .model_dump ())
926+
927+ # Build chat history with:
928+ # 1. Long-term memory (persistent context, broadcast messages)
929+ # 2. Short-term instance messages (current workflow specific)
930+ # 3. Additional context memory (for broadcast-triggered workflows)
853931 chat_history = []
932+
933+ # Add long-term memory first (broadcast messages, persistent context)
934+ chat_history .extend (long_term_memory_messages )
935+
936+ # Add short-term instance messages (current workflow)
854937 for msg in instance_messages :
855938 if isinstance (msg , dict ):
856939 chat_history .append (msg )
@@ -860,6 +943,9 @@ def _construct_messages_with_instance_history(
860943 msg .model_dump () if hasattr (msg , "model_dump" ) else dict (msg )
861944 )
862945
946+ # Add additional context memory last (for broadcast-triggered workflows)
947+ chat_history .extend (additional_context_messages )
948+
863949 if isinstance (input_data , str ):
864950 formatted_messages = self .prompt_template .format_prompt (
865951 chat_history = chat_history
0 commit comments