[BREAKING] feat: add Dapr integration (#3)

* feat: add Dapr integration
refactor: stop callback handling in Team class to accept whole conversation [BREAKING]

* update: README

* fix: dapr actors

* fix: test case causing issues
add: docs for remote and dapr

* fix: unit test

Author: unsafecode

.vscode/settings.json

@@ -4,7 +4,7 @@
     "editor.formatOnSave": true,
     "[python]": {
         "editor.codeActionsOnSave": {
-            "source.organizeImports": true,
+            "source.organizeImports": "explicit",
             "source.unusedImports": "always"
         }
     }

README.md

@@ -44,6 +44,14 @@ This project framework provides the following features:
 - Multiple strategies for agent to filter conversation messages (All, last N, top K and Last N, summarize, etc..)
 - LLMLingua (`extras` module) support to compress system prompts via strategies
 - LLM support for Structured Output
+- DAPR integration
+  - Native submodule to host `Askable` and `Workflow` as Dapr Actors
+  - Dapr _PubSub_ integration for `Workflow` to enable
+    - Event sourcing
+    - Decoupled communication between workflows
+- Multi-agent chat with multiple users
+  - Dapr PubSub integrations allows to move from one-to-many to many-to-many conversations, with different `User` instances impersonating different user profiles
+  - Demo Repo (_Coming soon_)
 - Remoting support ((`remote` module)), allowing agents to be run on a remote server and accessed elsewhere
   - REST and gRPC channels supported
   - Default implementation to run hosts with agent discovery and registration
@@ -56,8 +64,6 @@ This project framework provides the following features:
   - Azure AI Search plugin
   - DB plugin
   - API plugin
-- DAPR integration
-- Multi-agent chat with multiple users
 
 ## Getting Started
 
@@ -131,6 +137,22 @@ result = workflow.run("Hello, I'd like to know more about your products.")
 print(workflow.conversation.messages)
 ```
 
+### Submodules
+
+#### `remote`
+
+This module provides a way to run agents on a remote server and access them elsewhere. It includes support for REST and gRPC channels, as well as a default implementation to run hosts with agent discovery and registration.
+
+Additionally, it features Dapr integration, allowing for the hosting of `Askable` and `Workflow` as Dapr Actors, enabling event sourcing and decoupled communication between workflows.
+
+See the [Remote Agents documentation](docs/remote.md) and [Actors documentation](docs/actors.md) for more information.
+
+#### `extras`
+
+This module provides additional features to enhance the functionality of the framework. It includes support for:
+
+- `LLMLingua` to compress system prompts
+
 ## Demos
 
 `notebooks` folder contains a few demo notebooks that demonstrate how to use the framework in various scenarios.

docs/actors.md

@@ -0,0 +1,66 @@
+# Workflows as Actors
+
+Vanilla also offers a Virtual Actor pattern implementation via Dapr. This allows for the hosting of `Workflow` as Dapr Actors, enabling event sourcing and decoupled communication between workflows, along with at-most-one client-to-actor interactions - only one client can interact with an actor at a time.
+
+Being more specific, accessing an actor is done through a unique ID, which is used to identify the actor instance, that in our case matches a `Workflow` instance with its `Conversation`.
+
+## Dapr integration
+
+Dapr is a portable, event-driven runtime that makes it easy for developers to build resilient, microservice stateful applications. It offers a Virtual Actor pattern implementation, which allows for the hosting of `Workflow` as [Actors](https://docs.dapr.io/developing-applications/building-blocks/actors/actors-overview/) enabling event sourcing and decoupled communication between workflows via [PubSub](https://docs.dapr.io/developing-applications/building-blocks/pubsub/).
+
+Thanks to Dapr components, Vanilla is independent of the underlying infrastructure, and can run on any platform that supports Dapr, such as `Azure Cosmos DB` for state management and `Azure Service Bus` for PubSub.
+
+## Comparing to Remote Agents
+
+The key difference between is standard remoting simply offers **stateless** agents, while Dapr Actors offer **stateful** workflows. This means that Dapr Actors can maintain state across multiple interactions, while remote agents are simply APIs that are called and return a result.
+
+See the [Remote Agents documentation](remote.md) for more information.
+
+## Specification
+
+`vanilla_aiagents.remote.actors` provides the `WorkflowActor` class, which is simple wrapper around a `Workflow` instance with these main methods:
+
+- `run`: Runs the `Workflow` instance with the configured `Askable` over the actor's `Conversation` state.
+- `run_stream`: Runs the `Workflow` instance with the configured `Askable` over the actor's `Conversation` state, but in streaming outputs as PubSub events.
+- `get_conversation`: Gets the current state of the `Conversation` in the actor.
+
+Additionally, `vanilla_aiagents.remote.run_actors` provides a simple entrypoint to run a Dapr actor host with the `WorkflowActor` instances, and implements a PubSub event handler to receive `input` events to relay to the actors.
+
+## Diagram
+
+When a client calls an actor method (for example, `WorkflowActor.run`), Dapr locates the correct `WorkflowActor` by its unique ID and forwards the request. The actor then runs the `Workflow` with the provided input and can publish interim events to PubSub (if using `run_stream`). Finally, the actor sends its result back to the Dapr runtime, which returns it to the client.
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant DaprRuntime
+    participant WorkflowActor
+    participant PubSub
+
+    Client->>DaprRuntime: Invoke actor method (e.g. run, run_stream)
+    DaprRuntime->>WorkflowActor: Dispatch request by actor ID
+    WorkflowActor->>Workflow: Execute Askable logic
+    WorkflowActor->>PubSub: Publish events (e.g. stream data or stop signal)
+    WorkflowActor->>DaprRuntime: Return response
+    DaprRuntime->>Client: Response from actor
+```
+
+## Events format
+
+All events are sent as JSON strings following Dapr's PubSub conventions, leveraging CloudEvent v1 format. `vanilla_aiagents.remote.actors` provides a `WorkflowEvent` class to help with event serialization and deserialization, along with proper subclasses for each event type.
+
+All events have the following attributes:
+
+- `id`: The conversation ID, which is unique to the `Workflow` instance.
+- `type`: The event type, which can be `input`, `stream`, `update` or `stop`.
+
+Specific event types have additional attributes:
+
+- `input`: The input event, which is the input to the `Workflow` instance.
+  - `input`: The input data to the `Workflow` instance.
+- `stream`: The stream event, which is the partial output of the `Workflow` instance.
+  - `mark`: The mark of the stream event.
+  - `content`: The partial output of the `Workflow` instance.
+- `stop`: The stop event, which is the signal to stop the `Workflow` instance.
+  - `source`: The source `Askable` of the stop signal.
+- `update`: The update event, which is the signal to update the `Conversation` state.

docs/remote.md

@@ -0,0 +1,54 @@
+# Remote agents
+
+The framework supports remote agents. This allows you to run agents on a remote host and interact with them from another one, fostering a more distributed architecture and enabling agent reuse across different applications and `Team`.
+
+The `remote` submodule provides a default implementation to run hosts with agent discovery and registration. It supports REST and gRPC channels. Then, you can simply swap any `Agent` with a `RemoteAgent` and interact with it as if it were local.
+
+## Comparing to Actors
+
+Vanilla also offers a Virtual Actor pattern implementation via Dapr. This allows for the hosting of `Workflow` as Dapr Actors, enabling event sourcing and decoupled communication between workflows.
+
+The key difference between is standard remoting simply offers **stateless** agents, while Dapr Actors offer **stateful** workflows. This means that Dapr Actors can maintain state across multiple interactions, while remote agents are simply APIs that are called and return a result.
+
+See the [Actors documentation](actors.md) for more information.
+
+## Example
+
+```python
+# agent_entry.py
+
+from vanilla_aiagents.agent import Agent
+
+agent = Agent(id="agent", description="A remote agent", system_message="You are a remote agent.")
+```
+
+```dockerfile
+# Dockerfile
+
+FROM python:3.12-slim
+
+# Do all the copying and installing
+# ...
+
+# Expose the application port (must match the one specified below)
+EXPOSE 80
+
+# This is the entrypoint, Vanilla will automatically look for "_entry.py" files and expose them as agents
+CMD ["python", "-m", "vanilla_aiagents.remote.run_host", "--source-dir", ".", "--type", "rest", "--host", "0.0.0.0", "--port", "80"]
+```
+
+```python
+# client.py
+
+from vanilla_aiagents.remote.remote import RemoteAskable, RESTConnection
+
+# Create a remote connection with the given protocol and host
+# NOTE: Make sure to set the HOST_URL environment variable to the host URL
+remote_connection = RESTConnection(url=os.getenv("HOST_URL"))
+
+# ID here must match the agent ID in the remote host to use
+# NOTE: remote must be running and reachable at this point, since the RemoteAskable will try to connect to it
+remote = RemoteAskable(id="agent", connection=remote_connection)
+
+team = Team(id="team", description="Contoso team", members=[remote], llm=llm)
+```

notebooks/basic_user.ipynb

@@ -94,7 +94,7 @@
    "outputs": [],
    "source": [
     "# Assemble the team and workflow\n",
-    "flow = Team(id=\"team\", description=\"\", members=[agent1, user], llm=llm, stop_callback=lambda msgs: len(msgs) > 6)\n",
+    "flow = Team(id=\"team\", description=\"\", members=[agent1, user], llm=llm, stop_callback=lambda conv: len(conv.messages) > 6)\n",
     "workflow = Workflow(askable=flow)"
    ]
   },

notebooks/coding_azure.ipynb

@@ -76,7 +76,7 @@
     "from vanilla_aiagents.azure_coding_agent import AzureCodingAgent\n",
     "agent1 = AzureCodingAgent(id=\"agent1\", description=\"Agent 1\", llm=llm)\n",
     "\n",
-    "flow = Team(id=\"team\", description=\"An agent capable of writing and executing code\", members=[agent1], llm=llm, stop_callback=lambda msgs: len(msgs) > 2)\n",
+    "flow = Team(id=\"team\", description=\"An agent capable of writing and executing code\", members=[agent1], llm=llm, stop_callback=lambda conv: len(conv.messages) > 2)\n",
     "workflow = Workflow(askable=flow)\n",
     "workflow.conversation.variables[\"conversation_id\"] = \"test\"\n",
     "\n",

notebooks/coding_local.ipynb

@@ -70,7 +70,7 @@
     "from vanilla_aiagents.coding_agent import LocalCodingAgent\n",
     "agent1 = LocalCodingAgent(id=\"agent1\", description=\"Agent 1\", llm=llm)\n",
     "\n",
-    "flow = Team(id=\"team\", description=\"An agent capable of writing and executing code\", members=[agent1], llm=llm, stop_callback=lambda msgs: len(msgs) > 2)\n",
+    "flow = Team(id=\"team\", description=\"An agent capable of writing and executing code\", members=[agent1], llm=llm, stop_callback=lambda conv: len(conv.messages) > 2)\n",
     "workflow = Workflow(askable=flow)\n",
     "\n",
     "workflow.run(\"What is latest quote for Apple Inc. stock? Generate code to get the quote.\")\n",

notebooks/include_tools.ipynb

@@ -90,7 +90,7 @@
     "    return \"User balance: $100\"\n",
     "\n",
     "# include_tools_descriptions is set to True\n",
-    "flow = Team(id=\"flow\", description=\"\", members=[first, second], llm=llm, stop_callback=lambda x: len(x) > 2, include_tools_descriptions=True, use_structured_output=False)\n"
+    "flow = Team(id=\"flow\", description=\"\", members=[first, second], llm=llm, stop_callback=lambda conv: len(conv.messages) > 2, include_tools_descriptions=True, use_structured_output=False)\n"
    ]
   },
   {

notebooks/nested.ipynb

@@ -146,15 +146,15 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 7,
+   "execution_count": null,
    "metadata": {},
    "outputs": [],
    "source": [
     "step1 = Team(id=\"step1\", description=\"Validates the input provided to ensure all required information are collected\", \n",
     "             members=[data_collector, data_evaluator], \n",
     "             system_prompt=\"\",\n",
     "             llm=llm, \n",
-    "             stop_callback=lambda messages: messages[-1]['content'].lower().endswith(\"done\"))"
+    "             stop_callback=lambda conv: conv.messages[-1]['content'].lower().endswith(\"done\"))"
    ]
   },
   {
@@ -185,12 +185,12 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 10,
+   "execution_count": null,
    "metadata": {},
    "outputs": [],
    "source": [
     "step2 = Team(id=\"step2\", description=\"Processes the information gathered to provide the final answer\", members=[approver, approval_manager], llm=llm, \n",
-    "    stop_callback=lambda messages: messages[-1]['content'].lower().endswith(\"approve\") or messages[-1]['content'].lower().endswith(\"reject\"))"
+    "    stop_callback=lambda conv: conv.messages[-1]['content'].lower().endswith(\"approve\") or conv.messages[-1]['content'].lower().endswith(\"reject\"))"
    ]
   },
   {

notebooks/streaming.ipynb

@@ -150,7 +150,7 @@
     "\n",
     "user = User(id=\"user\", mode=\"unattended\")\n",
     "\n",
-    "flow = Team(id=\"team\", description=\"\", members=[agent1, user], llm=llm, stop_callback=lambda msgs: len(msgs) > 4)\n",
+    "flow = Team(id=\"team\", description=\"\", members=[agent1, user], llm=llm, stop_callback=lambda conv: len(conv.messages) > 4)\n",
     "\n"
    ]
   },

notebooks/telco_callcenter.ipynb

@@ -254,7 +254,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "team = Team(id=\"team\", description=\"\", members=[user, greeter, sales, technical, legal], llm=llm, stop_callback=lambda msgs: len(msgs) > 6)\n",
+    "team = Team(id=\"team\", description=\"\", members=[user, greeter, sales, technical, legal], llm=llm, stop_callback=lambda conv: len(conv.messages) > 6)\n",
     "workflow = Workflow(askable=team)"
    ]
   },

notebooks/telco_callcenter_sequence.ipynb

@@ -195,7 +195,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "team = Team(id=\"team\", description=\"\", members=[user, greeter, sales, technical], llm=llm, stop_callback=lambda msgs: len(msgs) > 6)\n"
+    "team = Team(id=\"team\", description=\"\", members=[user, greeter, sales, technical], llm=llm, stop_callback=lambda conv: len(conv.messages) > 6)\n"
    ]
   },
   {

requirements.txt

@@ -1,21 +1,23 @@
-openai==1.54.0
-httpx==0.27.2
-python-dotenv==1.0.1
-invoke==2.2.0
-pydantic==2.9.1
-azure-identity==1.19.0
-pytest==8.3.3
-coverage==7.6.1
-pytest-cov==5.0.0
-fastapi==0.115.5
-uvicorn==0.32.0
-grpcio-tools==1.66.2
-grpcio-reflection==1.67.0
-starlette_gzip_request==0.1.0
-pdoc==15.0.0
-llmlingua==0.2.2
-pydocstyle==6.3.0
-flake8==7.1.1
-flake8-docstrings==1.7.0
-docformatter==1.7.5
-black===24.10.0
+openai>=1.54.0
+httpx>=0.27.2
+python-dotenv>=1.0.1
+invoke>=2.2.0
+pydantic>=2.9.1
+azure-identity>=1.19.0
+pytest>=8.3.3
+coverage>=7.6.1
+pytest-cov>=5.0.0
+fastapi>=0.115.5
+uvicorn>=0.32.0
+grpcio-tools>=1.66.2
+grpcio-reflection>=1.67.0
+starlette_gzip_request>=0.1.0
+pdoc>=15.0.0
+llmlingua>=0.2.2
+pydocstyle>=6.3.0
+flake8>=7.1.1
+flake8-docstrings>=1.7.0
+docformatter>=1.7.5
+black>=24.10.0
+cloudevents>=1.11.0
+dapr-ext-fastapi>=1.14.0

samples/remote/actor/_actor_askable.py

@@ -0,0 +1,33 @@
+from typing import Annotated
+import os
+from vanilla_aiagents.agent import Agent
+from vanilla_aiagents.llm import AzureOpenAILLM
+
+llm = AzureOpenAILLM(
+    {
+        "azure_deployment": os.getenv("AZURE_OPENAI_MODEL"),
+        "azure_endpoint": os.getenv("AZURE_OPENAI_ENDPOINT"),
+        "api_key": os.getenv("AZURE_OPENAI_KEY"),
+        "api_version": os.getenv("AZURE_OPENAI_API_VERSION"),
+    }
+)
+
+_actor_askable = Agent(
+    id="agent",
+    llm=llm,
+    description="Call this agent to play guess the number game with the use",
+    system_message="""You are an AI assistant
+        Your task is to play a game with the user.
+        You first generate a random number between 1 and 100. Then save it as a conversation variable named "number".
+        The user will try to guess the number.
+        If the user's guess is too high, respond with "Too high".
+        If the user's guess is too low, respond with "Too low".
+        """,
+)
+
+
+@_actor_askable.register_tool(description="Generate a random number")
+def random() -> Annotated[str, "A random number"]:
+    import random
+
+    return str(random.randint(1, 100))

samples/remote/actor/run.py

@@ -0,0 +1,12 @@
+import sys
+import os
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+sys.path.append(os.path.abspath(os.path.join("../../../vanilla_aiagents")))
+
+from vanilla_aiagents.remote.dapr.run_actors import main
+
+if __name__ == "__main__":
+    main()