feat: add readme + imgs + todos + cleanup (#136)

* feat: add readme + imgs + todos + cleanup

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

* style: lint things

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

* fix: address feedback

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

* style: rm comment

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

* style: update comments

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

* fix(build): tox -e ruff

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

---------

Signed-off-by: Samantha Coyle <[email protected]>
Co-authored-by: Yaron Schneider <[email protected]>

Author: sicoyle

dapr_agents/agent/actor/base.py

@@ -38,16 +38,16 @@ async def _on_activate(self) -> None:
         if not has_state:
             # Initialize state with default values if it doesn't exist
             logger.info(f"Initializing state for {self.actor_id}")
-            self.state = AgentActorState(overall_status=AgentStatus.IDLE)
+            self.agent_state = AgentActorState(status=AgentStatus.IDLE)
             await self._state_manager.set_state(
-                self.agent_state_key, self.state.model_dump()
+                self.agent_state_key, self.agent_state.model_dump()
             )
             await self._state_manager.save_state()
         else:
             # Load existing state
             logger.info(f"Loading existing state for {self.actor_id}")
             logger.debug(f"Existing state for {self.actor_id}: {state_data}")
-            self.state = AgentActorState(**state_data)
+            self.agent_state = AgentActorState(**state_data)
 
     async def _on_deactivate(self) -> None:
         """
@@ -61,9 +61,9 @@ async def set_status(self, status: AgentStatus) -> None:
         """
         Sets the current operational status of the agent and saves the state.
         """
-        self.state.overall_status = status
+        self.agent_state.status = status
         await self._state_manager.set_state(
-            self.agent_state_key, self.state.model_dump()
+            self.agent_state_key, self.agent_state.model_dump()
         )
         await self._state_manager.save_state()
 
@@ -101,11 +101,11 @@ async def invoke_task(self, task: Optional[str] = None) -> str:
             input=task_entry_input,
             status=AgentTaskStatus.IN_PROGRESS,
         )
-        self.state.task_history.append(task_entry)
+        self.agent_state.task_history.append(task_entry)
 
         # Save initial task state with IN_PROGRESS status
         await self._state_manager.set_state(
-            self.agent_state_key, self.state.model_dump()
+            self.agent_state_key, self.agent_state.model_dump()
         )
         await self._state_manager.save_state()
 
@@ -134,7 +134,7 @@ async def invoke_task(self, task: Optional[str] = None) -> str:
         finally:
             # Ensure the final state of the task is saved
             await self._state_manager.set_state(
-                self.agent_state_key, self.state.model_dump()
+                self.agent_state_key, self.agent_state.model_dump()
             )
             await self._state_manager.save_state()
             # Revert the agent's status to idle
@@ -152,12 +152,12 @@ async def add_message(self, message: Union[AgentActorMessage, dict]) -> None:
             message = AgentActorMessage(**message)
 
         # Add the new message to the state
-        self.state.messages.append(message)
-        self.state.message_count += 1
+        self.agent_state.messages.append(message)
+        self.agent_state.message_count += 1
 
         # Save state back to Dapr
         await self._state_manager.set_state(
-            self.agent_state_key, self.state.model_dump()
+            self.agent_state_key, self.agent_state.model_dump()
         )
         await self._state_manager.save_state()
 
@@ -178,7 +178,6 @@ async def get_messages(self) -> List[dict]:
                 # Return the list of messages as dictionaries (timestamp will be automatically serialized to ISO format)
                 return [message.model_dump() for message in state.messages]
             except ValidationError as e:
-                # Handle validation errors
                 print(f"Validation error: {e}")
                 return []
         return []

dapr_agents/agent/patterns/react/base.py

@@ -36,6 +36,8 @@ class ReActAgent(AgentBase):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
+    # TODO: build this from a template file
+    # TODO: tool for evaluating prompt updates here and catching err drift
     def construct_system_prompt(self) -> str:
         """
         Constructs a system prompt in the ReAct reasoning-action format based on the agent's attributes and tools.
@@ -54,7 +56,6 @@ def construct_system_prompt(self) -> str:
         prompt_parts.append("## Role\nYour role is {{role}}.")
         prompt_parts.append("## Goal\n{{goal}}.")
 
-        # Append instructions if provided
         if self.instructions:
             prompt_parts.append("## Instructions\n{{instructions}}")
 
@@ -68,7 +69,7 @@ def construct_system_prompt(self) -> str:
             tools_section.rstrip()
         )  # Trim any trailing newlines from tools_section
 
-        # Additional Guidelines
+        # TODO: move this to a separate file
         additional_guidelines = textwrap.dedent(
             """
         If you think about using tool, it must use the correct tool JSON blob format as shown below:
@@ -83,6 +84,7 @@ def construct_system_prompt(self) -> str:
         prompt_parts.append(additional_guidelines)
 
         # ReAct specific guidelines
+        # TODO: move this to a separate file
         react_guidelines = textwrap.dedent(
             """
         ## ReAct Format
@@ -157,7 +159,6 @@ async def run(self, input_data: Optional[Union[str, Dict[str, Any]]] = None) ->
         if user_message:
             self.text_formatter.print_message(user_message)
 
-        # Get Tool Names to validate tool selection
         available_tools = self.tool_executor.get_tool_names()
 
         # Initialize react_loop for iterative reasoning
@@ -215,12 +216,11 @@ async def run(self, input_data: Optional[Union[str, Dict[str, Any]]] = None) ->
                         "No action specified; continuing with further reasoning."
                     )
                     react_loop += f"Thought:{thought_action}\n"
-                    continue  # Proceed to the next iteration
+                    continue
 
                 action_name = action["name"]
                 action_args = action["arguments"]
 
-                # Print Action
                 self.text_formatter.print_react_part("Action", json.dumps(action))
 
                 if action_name not in available_tools:
@@ -229,7 +229,6 @@ async def run(self, input_data: Optional[Union[str, Dict[str, Any]]] = None) ->
                 logger.info(f"Executing {action_name} with arguments {action_args}")
                 result = await self.tool_executor.run_tool(action_name, **action_args)
 
-                # Print Observation
                 self.text_formatter.print_react_part("Observation", result)
                 react_loop += f"Thought:{thought_action}\nAction:{json.dumps(action)}\nObservation:{result}\n"
 

dapr_agents/service/fastapi/base.py

@@ -58,7 +58,6 @@ def model_post_init(self, __context: Any) -> None:
             lifespan=self.lifespan,
         )
 
-        # Configure CORS settings
         self.app.add_middleware(
             CORSMiddleware,
             allow_origins=self.cors_origins,

dapr_agents/types/agent.py

@@ -78,6 +78,6 @@ class AgentActorState(BaseModel):
     task_history: Optional[List[AgentTaskEntry]] = Field(
         default_factory=list, description="History of tasks the agent has performed"
     )
-    overall_status: AgentStatus = Field(
+    status: AgentStatus = Field(
         AgentStatus.IDLE, description="Current operational status of the agent"
     )

dapr_agents/workflow/agentic.py

@@ -74,6 +74,7 @@ class AgenticWorkflow(WorkflowApp, DaprPubSub, MessageRoutingMixin):
     agents_registry_key: str = Field(
         default="agents_registry", description="Key for agents registry in state store."
     )
+    # TODO: test this is respected by runtime.
     max_iterations: int = Field(
         default=20, description="Maximum iterations for workflows.", ge=1
     )
@@ -109,6 +110,7 @@ def model_post_init(self, __context: Any) -> None:
         self._text_formatter = ColorTextFormatter()
 
         # Initialize state store client (used for persisting workflow state to Dapr)
+        # Why make a state store client and dapr_client here?
         self._state_store_client = DaprStateStore(store_name=self.state_store_name)
         logger.info(f"State store '{self.state_store_name}' initialized.")
 

dapr_agents/workflow/base.py

@@ -44,6 +44,7 @@ class WorkflowApp(BaseModel):
     llm: Optional[ChatClientBase] = Field(
         default=None, description="The default LLM client for all LLM-based tasks."
     )
+    # TODO: I think this should be within the wf client or wf runtime...?
     timeout: int = Field(
         default=300,
         description="Default timeout duration in seconds for workflow tasks.",
@@ -181,6 +182,21 @@ def wrapper(ctx: WorkflowActivityContext, *args, **kwargs):
 
         return wrapper
 
+    # TODO: workflow discovery can also come from dapr runtime
+    # Python workflows can be registered in a variety of ways, and we need to support all of them.
+    # This supports decorator-based registration;
+    # however, there is also manual registration approach.
+    # See example below:
+    #     def setup_workflow_runtime():
+    #       wf_runtime = WorkflowRuntime()
+    #       wf_runtime.register_workflow(order_processing_workflow)
+    #       wf_runtime.register_workflow(fulfillment_workflow)
+    #       wf_runtime.register_activity(process_payment)
+    #       wf_runtime.register_activity(send_notification)
+    #     return wf_runtime
+
+    # runtime = setup_workflow_runtime()
+    # runtime.start()
     def _discover_workflows(self) -> Dict[str, Callable]:
         """Gather all @workflow-decorated functions and methods."""
         module = sys.modules["__main__"]
@@ -236,7 +252,8 @@ def register_agent(
             key (str): The key to update.
             data (dict): The data to update the store with.
         """
-        # retry the entire operation up to ten times sleeping 1 second between each attempt
+        # retry the entire operation up to ten times sleeping 1 second between each
+        # TODO: rm the custom retry logic here and use the DaprClient retry_policy instead.
         for attempt in range(1, 11):
             try:
                 response: StateResponse = self.client.get_state(

dapr_agents/workflow/orchestrators/README.md

@@ -0,0 +1,10 @@
+# Orchestrators
+
+Available Workflow options to orchestrate communication between agents:
+
+- LLM-based: Uses a large language model (e.g., GPT-4o) to determine the most suitable agent based on the message and context.
+- Random: Selects an agent randomly for each task.
+- RoundRobin: Cycles through agents in a fixed order, ensuring each agent gets an equal opportunity to process tasks.
+
+## Visual representation of each orchestration option:
+![Orchestrator workflows visualized](./orchestratorWorkflows.png)

dapr_agents/workflow/orchestrators/llm/orchestrator.py

@@ -60,6 +60,9 @@ def model_post_init(self, __context: Any) -> None:
 
     @message_router
     @workflow(name="LLMWorkflow")
+    # TODO: set retry policies on the activities!
+    # TODO: utilize prompt verdict value of failed as we do not currently use.
+    # https://github.com/dapr/dapr-agents/pull/136#discussion_r2175751545
     def main_workflow(self, ctx: DaprWorkflowContext, message: TriggerAction):
         """
         Executes an LLM-driven agentic workflow where the next agent is dynamically selected
@@ -710,6 +713,7 @@ async def finish_workflow(
         # Store the final summary and verdict in workflow state
         await self.update_workflow_state(instance_id=instance_id, final_output=summary)
 
+    # TODO: this should be a compensating activity called in the event of an error from any other activity.
     async def update_workflow_state(
         self,
         instance_id: str,