fix: decouple memory + state to work with one and/or both (#236)

* fix: decouple memory + state to work with one and/or both for agent+orchestrator

Signed-off-by: Samantha Coyle <[email protected]>

* feat: add memory on llm orchestrator quickstart

Signed-off-by: Samantha Coyle <[email protected]>

* style: tox -e ruff

Signed-off-by: Samantha Coyle <[email protected]>

---------

Signed-off-by: Samantha Coyle <[email protected]>

Author: sicoyle

dapr_agents/agents/durableagent/agent.py

@@ -60,10 +60,8 @@ class DurableAgent(AgenticWorkflow, AgentBase):
         default=None,
         description="The current workflow instance ID for this agent.",
     )
-    memory: ConversationDaprStateMemory = Field(
-        default_factory=lambda: ConversationDaprStateMemory(
-            store_name="workflowstatestore", session_id="durable_agent_session"
-        ),
+    memory: Optional[ConversationDaprStateMemory] = Field(
+        default=None,
         description="Persistent memory with session-based state hydration.",
     )
 
@@ -93,19 +91,31 @@ def model_post_init(self, __context: Any) -> None:
         if not self.state:
             self.state = {"instances": {}}
 
+        if self.memory is not None:
+            self.memory = ConversationDaprStateMemory(
+                store_name=self.memory.store_name,
+                session_id=f"{self.name or 'agent'}_session",
+            )
+            logger.info(f"Initialized memory with store name: {self.memory.store_name}")
+
         # Load the current workflow instance ID from state using session_id
         logger.debug(f"State after loading: {self.state}")
         if self.state and self.state.get("instances"):
             logger.debug(f"Found {len(self.state['instances'])} instances in state")
+            current_session_id = (
+                self.memory.session_id
+                if self.memory
+                else f"{self.name}_default_session"
+            )
             for instance_id, instance_data in self.state["instances"].items():
                 stored_workflow_name = instance_data.get("workflow_name")
                 stored_session_id = instance_data.get("session_id")
                 logger.debug(
-                    f"Instance {instance_id}: workflow_name={stored_workflow_name}, session_id={stored_session_id}, current_workflow_name={self._workflow_name}, current_session_id={self.memory.session_id}"
+                    f"Instance {instance_id}: workflow_name={stored_workflow_name}, session_id={stored_session_id}, current_workflow_name={self._workflow_name}, current_session_id={current_session_id}"
                 )
                 if (
                     stored_workflow_name == self._workflow_name
-                    and stored_session_id == self.memory.session_id
+                    and stored_session_id == current_session_id
                 ):
                     self.workflow_instance_id = instance_id
                     logger.debug(
@@ -401,7 +411,9 @@ def record_initial_entry(
             "workflow_instance_id": instance_id,
             "triggering_workflow_instance_id": triggering_workflow_instance_id,
             "workflow_name": self._workflow_name,
-            "session_id": self.memory.session_id,
+            "session_id": self.memory.session_id
+            if self.memory
+            else f"{self.name}_default_session",
             "start_time": start_time_str,
             "trace_context": trace_context,
             "status": DaprWorkflowStatus.RUNNING.value,
@@ -443,7 +455,9 @@ def _ensure_instance_exists(
                 "workflow_instance_id": instance_id,
                 "triggering_workflow_instance_id": triggering_workflow_instance_id,
                 "workflow_name": self._workflow_name,
-                "session_id": self.memory.session_id,
+                "session_id": self.memory.session_id
+                if self.memory
+                else f"{self.name}_default_session",
                 "messages": [],
                 "tool_history": [],
                 "status": DaprWorkflowStatus.RUNNING.value,
@@ -462,7 +476,8 @@ def _process_user_message(
             return
 
         user_msg = UserMessage(content=user_message_copy.get("content", ""))
-        self.memory.add_message(user_msg)
+        if self.memory:
+            self.memory.add_message(user_msg)
 
         msg_object = DurableAgentMessage(**user_message_copy)
         inst = self.state["instances"][instance_id]
@@ -502,7 +517,8 @@ def _save_assistant_message(
         if not message_exists:
             messages_list.append(agent_msg.model_dump(mode="json"))
             inst["last_message"] = agent_msg.model_dump(mode="json")
-            self.memory.add_message(AssistantMessage(**assistant_message))
+            if self.memory:
+                self.memory.add_message(AssistantMessage(**assistant_message))
             self.save_state()
 
     def _print_llm_interaction_messages(
@@ -649,8 +665,9 @@ def _update_agent_memory_and_history(
         # Update tool history and memory of agent (only if new)
         # Note: Memory updates are handled at workflow level to avoid replay issues
         self.tool_history.append(tool_history_entry)
-        # Add the tool message to the agent's memory
-        self.memory.add_message(tool_message)
+        # Add the tool message to the agent's memory if enabled
+        if self.memory:
+            self.memory.add_message(tool_message)
 
     def _get_last_message_from_state(
         self, instance_id: str
@@ -850,8 +867,9 @@ async def process_broadcast_message(self, message: BroadcastMessage):
             logger.debug(
                 f"{self.name} processing broadcast message from '{source}'. Content: {message_content}"
             )
-            # Store the message in local memory
-            self.memory.add_message(message)
+            # Store the message in local memory if enabled
+            if self.memory:
+                self.memory.add_message(message)
 
             # Define DurableAgentMessage object for state persistence
             agent_msg = DurableAgentMessage(**message.model_dump())
@@ -920,15 +938,18 @@ def _construct_messages_with_instance_history(
         else:
             instance_messages = []
 
-        # Get messages from persistent memory (session-based, cross-workflow)
+        # Get messages from persistent memory (session-based, cross-workflow) if memory is enabled
         persistent_memory_messages = []
-        try:
-            persistent_memory_messages = self.memory.get_messages()
-            logger.info(
-                f"Retrieved {len(persistent_memory_messages)} messages for session {self.memory.session_id}"
-            )
-        except Exception as e:
-            logger.warning(f"Failed to retrieve persistent memory: {e}")
+        if self.memory:
+            try:
+                persistent_memory_messages = self.memory.get_messages()
+                logger.info(
+                    f"Retrieved {len(persistent_memory_messages)} messages for session {self.memory.session_id}"
+                )
+            except Exception as e:
+                logger.warning(f"Failed to retrieve persistent memory: {e}")
+        else:
+            logger.debug("Memory is disabled, skipping persistent memory retrieval")
 
         # Get long-term memory from workflow state (for broadcast messages and persistent context)
         long_term_memory_data = self.state.get("chat_history", [])

dapr_agents/memory/daprstatestore.py

@@ -89,7 +89,6 @@ def add_message(self, message: Union[Dict[str, Any], BaseMessage]) -> None:
         message_key = self._get_message_key(message_id)
         message.update(
             {
-                "sessionId": self.session_id,
                 "createdAt": datetime.now().isoformat() + "Z",
             }
         )

dapr_agents/workflow/orchestrators/llm/orchestrator.py

@@ -54,10 +54,8 @@ class LLMOrchestrator(OrchestratorWorkflowBase):
         default=None,
         description="The current workflow instance ID for this orchestrator.",
     )
-    memory: ConversationDaprStateMemory = Field(
-        default_factory=lambda: ConversationDaprStateMemory(
-            store_name="workflowstatestore", session_id="orchestrator_session"
-        ),
+    memory: Optional[ConversationDaprStateMemory] = Field(
+        default=None,
         description="Persistent memory with session-based state hydration.",
     )
 
@@ -77,14 +75,25 @@ def model_post_init(self, __context: Any) -> None:
         if not self.state:
             logger.debug("No state found, initializing empty state")
             self.state = {"instances": {}}
-        else:
-            logger.debug(f"State loaded successfully: {self.state}")
+
+        if self.memory is not None:
+            self.memory = ConversationDaprStateMemory(
+                store_name=self.memory.store_name,
+                session_id=f"{self.name or 'orchestrator'}_session",
+            )
+
+            print(f"sam memory store name is {self.memory.store_name}")
+            logger.info(f"sam memory store name is {self.memory.store_name}")
 
         # Load the current workflow instance ID from state using session_id)
         if self.state and self.state.get("instances"):
             logger.debug(f"Found {len(self.state['instances'])} instances in state")
 
-            current_session_id = self.memory.session_id
+            current_session_id = (
+                self.memory.session_id
+                if self.memory
+                else f"{self.name}_default_session"
+            )
             for instance_id, instance_data in self.state["instances"].items():
                 stored_workflow_name = instance_data.get("workflow_name")
                 stored_session_id = instance_data.get("session_id")
@@ -654,10 +663,15 @@ async def initialize_workflow_with_plan(
                 "instances", {}
             ).items():
                 stored_session_id = instance_data.get("session_id")
-                if stored_session_id == self.memory.session_id:
+                current_session_id = (
+                    self.memory.session_id
+                    if self.memory
+                    else f"{self.name}_default_session"
+                )
+                if stored_session_id == current_session_id:
                     existing_plan = instance_data.get("plan", [])
                     logger.debug(
-                        f"Found existing plan for session_id {self.memory.session_id} in instance {stored_instance_id}"
+                        f"Found existing plan for session_id {current_session_id} in instance {stored_instance_id}"
                     )
                     break
 
@@ -686,23 +700,36 @@ async def initialize_workflow_with_plan(
                     structured_mode="json",
                 )
 
-                # Parse the response - now we get a Pydantic model directly
-                if hasattr(response, "choices") and response.choices:
-                    # If it's still a raw response, parse it
-                    plan_data = response.choices[0].message.content
-                    logger.debug(f"Plan generation response: {plan_data}")
-                    plan_dict = json.loads(plan_data)
-                    # Convert raw dictionaries to Pydantic models
+                # Parse the response
+                if isinstance(response, str):
+                    # If it's a raw JSON string
+                    plan_dict = json.loads(response)
                     plan_objects = [
                         PlanStep(**step_dict)
                         for step_dict in plan_dict.get("objects", [])
                     ]
+                elif hasattr(response, "choices") and response.choices:
+                    # If it's an OpenAI-style response with multiple choices
+                    plan_objects = []
+                    for choice in response.choices:
+                        plan_data = choice.message.content
+                        if isinstance(plan_data, str):
+                            plan_dict = json.loads(plan_data)
+                            plan_objects.extend(
+                                PlanStep(**step_dict)
+                                for step_dict in plan_dict.get("objects", [])
+                            )
+                        elif hasattr(plan_data, "objects"):
+                            plan_objects.extend(plan_data.objects)
                 else:
                     # If it's already a Pydantic model
                     plan_objects = (
                         response.objects if hasattr(response, "objects") else []
                     )
-                    logger.debug(f"Plan generation response (Pydantic): {plan_objects}")
+
+                logger.debug(
+                    f"Plan generation response with {len(plan_objects)} objects: {plan_objects}"
+                )
 
             # Format and broadcast message
             plan_dicts = self._convert_plan_objects_to_dicts(plan_objects)
@@ -1030,8 +1057,9 @@ async def update_workflow_state(
             workflow_entry["messages"].append(serialized_message)
             workflow_entry["last_message"] = serialized_message
 
-            # Update the local chat history
-            self.memory.add_message(message)
+            # Update the local chat history if memory is enabled
+            if self.memory:
+                self.memory.add_message(message)
 
         if final_output is not None:
             workflow_entry["output"] = final_output
@@ -1041,7 +1069,9 @@ async def update_workflow_state(
         # Store workflow instance ID, workflow name, and session_id for session-based state rehydration
         workflow_entry["workflow_instance_id"] = instance_id
         workflow_entry["workflow_name"] = self._workflow_name
-        workflow_entry["session_id"] = self.memory.session_id
+        workflow_entry["session_id"] = (
+            self.memory.session_id if self.memory else f"{self.name}_default_session"
+        )
 
         # Persist updated state
         self.save_state()

quickstarts/03-agent-tool-call/weather_durable_agent_tracing.py

@@ -37,7 +37,6 @@ async def main():
         message_bus_name="messagepubsub",
         state_store_name="workflowstatestore",
         agents_registry_store_name="agentstatestore",
-        history_store_name="historystore",
     )
 
     await AIAgent.run("What is the weather in Virginia, New York and Washington DC?")

quickstarts/05-multi-agent-workflows/components/conversationmemory.yaml

@@ -0,0 +1,14 @@
+apiVersion: dapr.io/v1alpha1
+kind: Component
+metadata:
+  name: conversationstore
+spec:
+  type: state.redis
+  version: v1
+  metadata:
+    - name: redisHost
+      value: localhost:6379
+    - name: redisPassword
+      value: ""
+    - name: keyPrefix
+      value: none

quickstarts/05-multi-agent-workflows/services/workflow-llm/app.py

@@ -1,5 +1,6 @@
 from dapr_agents import LLMOrchestrator
 from dapr_agents.llm import DaprChatClient
+from dapr_agents.memory import ConversationDaprStateMemory
 from dotenv import load_dotenv
 import asyncio
 import logging
@@ -34,6 +35,9 @@ async def main():
             agents_registry_store_name="agentstatestore",
             agents_registry_key="agents_registry",
             broadcast_topic_name="beacon_channel",
+            memory=ConversationDaprStateMemory(
+                store_name="conversationstore", session_id="myuniqueid"
+            ),
             max_iterations=3,
         ).as_service(port=8004)