From 74bcd5e0f6f36508028e59d9ebad8e951d07f8ea Mon Sep 17 00:00:00 2001 From: peterychang <49209570+peterychang@users.noreply.github.com> Date: Tue, 26 Nov 2024 15:37:40 -0500 Subject: [PATCH 1/3] add protobuf serialization test (#4224) * add protobuf serialization test * proto file regeneration --------- Co-authored-by: Ryan Sweet Co-authored-by: Mohammad Mazraeh --- .github/workflows/checks.yml | 1 + python/packages/autogen-core/pyproject.toml | 6 +-- .../tests/protos/serialization_test.proto | 11 +++++ .../tests/protos/serialization_test_pb2.py | 28 +++++++++++ .../tests/protos/serialization_test_pb2.pyi | 46 +++++++++++++++++++ .../protos/serialization_test_pb2_grpc.py | 4 ++ .../protos/serialization_test_pb2_grpc.pyi | 17 +++++++ .../autogen-core/tests/test_serialization.py | 31 +++++++++++++ python/pyproject.toml | 2 + 9 files changed, 143 insertions(+), 3 deletions(-) create mode 100644 python/packages/autogen-core/tests/protos/serialization_test.proto create mode 100644 python/packages/autogen-core/tests/protos/serialization_test_pb2.py create mode 100644 python/packages/autogen-core/tests/protos/serialization_test_pb2.pyi create mode 100644 python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.py create mode 100644 python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.pyi diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index 31869b86ccc2..703c07494251 100644 --- a/.github/workflows/checks.yml +++ b/.github/workflows/checks.yml @@ -190,6 +190,7 @@ jobs: run: | source ${{ github.workspace }}/python/.venv/bin/activate poe gen-proto + poe gen-test-proto working-directory: ./python - name: Check if there are uncommited changes id: changes diff --git a/python/packages/autogen-core/pyproject.toml b/python/packages/autogen-core/pyproject.toml index 064928590b0f..8727f5ee0c25 100644 --- a/python/packages/autogen-core/pyproject.toml +++ b/python/packages/autogen-core/pyproject.toml @@ -81,7 +81,7 @@ dev-dependencies = [ [tool.ruff] extend = "../../pyproject.toml" -exclude = ["build", "dist", "src/autogen_core/application/protos"] +exclude = ["build", "dist", "src/autogen_core/application/protos", "tests/protos"] include = ["src/**", "samples/*.py", "docs/**/*.ipynb", "tests/**"] [tool.ruff.lint.per-file-ignores] @@ -91,7 +91,7 @@ include = ["src/**", "samples/*.py", "docs/**/*.ipynb", "tests/**"] [tool.pyright] extends = "../../pyproject.toml" include = ["src", "tests", "samples"] -exclude = ["src/autogen_core/application/protos"] +exclude = ["src/autogen_core/application/protos", "tests/protos"] reportDeprecated = false [tool.pytest.ini_options] @@ -111,7 +111,7 @@ include = "../../shared_tasks.toml" test = "pytest -n auto" mypy.default_item_type = "cmd" mypy.sequence = [ - "mypy --config-file ../../pyproject.toml --exclude src/autogen_core/application/protos src tests", + "mypy --config-file ../../pyproject.toml --exclude src/autogen_core/application/protos --exclude tests/protos src tests", "nbqa mypy docs/src --config-file ../../pyproject.toml", ] diff --git a/python/packages/autogen-core/tests/protos/serialization_test.proto b/python/packages/autogen-core/tests/protos/serialization_test.proto new file mode 100644 index 000000000000..611100ccde12 --- /dev/null +++ b/python/packages/autogen-core/tests/protos/serialization_test.proto @@ -0,0 +1,11 @@ +syntax = "proto3"; + +package agents; + +message ProtoMessage { + string message = 1; +} +message NestingProtoMessage { + string message = 1; + ProtoMessage nested = 2; +} \ No newline at end of file diff --git a/python/packages/autogen-core/tests/protos/serialization_test_pb2.py b/python/packages/autogen-core/tests/protos/serialization_test_pb2.py new file mode 100644 index 000000000000..ebc4bfee7018 --- /dev/null +++ b/python/packages/autogen-core/tests/protos/serialization_test_pb2.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: serialization_test.proto +# Protobuf Python Version: 4.25.1 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x18serialization_test.proto\x12\x06\x61gents\"\x1f\n\x0cProtoMessage\x12\x0f\n\x07message\x18\x01 \x01(\t\"L\n\x13NestingProtoMessage\x12\x0f\n\x07message\x18\x01 \x01(\t\x12$\n\x06nested\x18\x02 \x01(\x0b\x32\x14.agents.ProtoMessageb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'serialization_test_pb2', _globals) +if _descriptor._USE_C_DESCRIPTORS == False: + DESCRIPTOR._options = None + _globals['_PROTOMESSAGE']._serialized_start=36 + _globals['_PROTOMESSAGE']._serialized_end=67 + _globals['_NESTINGPROTOMESSAGE']._serialized_start=69 + _globals['_NESTINGPROTOMESSAGE']._serialized_end=145 +# @@protoc_insertion_point(module_scope) diff --git a/python/packages/autogen-core/tests/protos/serialization_test_pb2.pyi b/python/packages/autogen-core/tests/protos/serialization_test_pb2.pyi new file mode 100644 index 000000000000..b8a284663f6e --- /dev/null +++ b/python/packages/autogen-core/tests/protos/serialization_test_pb2.pyi @@ -0,0 +1,46 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import builtins +import google.protobuf.descriptor +import google.protobuf.message +import typing + +DESCRIPTOR: google.protobuf.descriptor.FileDescriptor + +@typing.final +class ProtoMessage(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + MESSAGE_FIELD_NUMBER: builtins.int + message: builtins.str + def __init__( + self, + *, + message: builtins.str = ..., + ) -> None: ... + def ClearField(self, field_name: typing.Literal["message", b"message"]) -> None: ... + +global___ProtoMessage = ProtoMessage + +@typing.final +class NestingProtoMessage(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + MESSAGE_FIELD_NUMBER: builtins.int + NESTED_FIELD_NUMBER: builtins.int + message: builtins.str + @property + def nested(self) -> global___ProtoMessage: ... + def __init__( + self, + *, + message: builtins.str = ..., + nested: global___ProtoMessage | None = ..., + ) -> None: ... + def HasField(self, field_name: typing.Literal["nested", b"nested"]) -> builtins.bool: ... + def ClearField(self, field_name: typing.Literal["message", b"message", "nested", b"nested"]) -> None: ... + +global___NestingProtoMessage = NestingProtoMessage diff --git a/python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.py b/python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.py new file mode 100644 index 000000000000..2daafffebfc8 --- /dev/null +++ b/python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.py @@ -0,0 +1,4 @@ +# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! +"""Client and server classes corresponding to protobuf-defined services.""" +import grpc + diff --git a/python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.pyi b/python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.pyi new file mode 100644 index 000000000000..a6a9cff9dfd4 --- /dev/null +++ b/python/packages/autogen-core/tests/protos/serialization_test_pb2_grpc.pyi @@ -0,0 +1,17 @@ +""" +@generated by mypy-protobuf. Do not edit manually! +isort:skip_file +""" + +import abc +import collections.abc +import grpc +import grpc.aio +import typing + +_T = typing.TypeVar("_T") + +class _MaybeAsyncIterator(collections.abc.AsyncIterator[_T], collections.abc.Iterator[_T], metaclass=abc.ABCMeta): ... + +class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext): # type: ignore[misc, type-arg] + ... diff --git a/python/packages/autogen-core/tests/test_serialization.py b/python/packages/autogen-core/tests/test_serialization.py index 3f3b0174c8b0..6b5568411f6f 100644 --- a/python/packages/autogen-core/tests/test_serialization.py +++ b/python/packages/autogen-core/tests/test_serialization.py @@ -11,6 +11,7 @@ from autogen_core.base._serialization import DataclassJsonMessageSerializer, PydanticJsonMessageSerializer from autogen_core.components import Image from PIL import Image as PILImage +from protos.serialization_test_pb2 import NestingProtoMessage, ProtoMessage from pydantic import BaseModel @@ -83,6 +84,36 @@ def test_nesting_dataclass_dataclass() -> None: serde.add_serializer(try_get_known_serializers_for_type(NestingDataclassMessage)) +def test_proto() -> None: + serde = SerializationRegistry() + serde.add_serializer(try_get_known_serializers_for_type(ProtoMessage)) + + message = ProtoMessage(message="hello") + name = serde.type_name(message) + # TODO: should be PROTO_DATA_CONTENT_TYPE + data = serde.serialize(message, type_name=name, data_content_type=JSON_DATA_CONTENT_TYPE) + assert name == "ProtoMessage" + # TODO: assert data == stuff + deserialized = serde.deserialize(data, type_name=name, data_content_type=JSON_DATA_CONTENT_TYPE) + assert deserialized == message + + +def test_nested_proto() -> None: + serde = SerializationRegistry() + serde.add_serializer(try_get_known_serializers_for_type(NestingProtoMessage)) + + message = NestingProtoMessage(message="hello", nested=ProtoMessage(message="world")) + name = serde.type_name(message) + + # TODO: should be PROTO_DATA_CONTENT_TYPE + data = serde.serialize(message, type_name=name, data_content_type=JSON_DATA_CONTENT_TYPE) + + # TODO: assert data == stuff + + deserialized = serde.deserialize(data, type_name=name, data_content_type=JSON_DATA_CONTENT_TYPE) + assert deserialized == message + + @dataclass class DataclassNestedUnionSyntaxOldMessage: message: Union[str, int] diff --git a/python/pyproject.toml b/python/pyproject.toml index 3b099db535e8..e9b9753cfca9 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -79,3 +79,5 @@ test = "python run_task_in_pkgs_if_exist.py test" check = ["fmt", "lint", "pyright", "mypy", "test"] gen-proto = "python -m grpc_tools.protoc --python_out=./packages/autogen-core/src/autogen_core/application/protos --grpc_python_out=./packages/autogen-core/src/autogen_core/application/protos --mypy_out=./packages/autogen-core/src/autogen_core/application/protos --mypy_grpc_out=./packages/autogen-core/src/autogen_core/application/protos --proto_path ../protos/ agent_worker.proto --proto_path ../protos/ cloudevent.proto" + +gen-test-proto = "python -m grpc_tools.protoc --python_out=./packages/autogen-core/tests/protos --grpc_python_out=./packages/autogen-core/tests/protos --mypy_out=./packages/autogen-core/tests/protos --mypy_grpc_out=./packages/autogen-core/tests/protos --proto_path ./packages/autogen-core/tests/protos serialization_test.proto" \ No newline at end of file From 1f07e5bea5e776f53a76fba9ba78258e5cb6aa18 Mon Sep 17 00:00:00 2001 From: Jack Gerrits Date: Tue, 26 Nov 2024 15:58:26 -0500 Subject: [PATCH 2/3] Add type prefix subscription (#4383) * Add type prefix subscription * update example --------- Co-authored-by: Ryan Sweet --- docs/design/04 - Agent and Topic ID Specs.md | 2 +- protos/agent_worker.proto | 6 ++ .../application/_worker_runtime.py | 47 +++++++++----- .../_worker_runtime_host_servicer.py | 50 +++++++++------ .../application/protos/agent_worker_pb2.py | 36 ++++++----- .../application/protos/agent_worker_pb2.pyi | 28 ++++++++- .../src/autogen_core/components/__init__.py | 2 + .../components/_type_prefix_subscription.py | 63 +++++++++++++++++++ .../components/_type_subscription.py | 6 +- 9 files changed, 181 insertions(+), 59 deletions(-) create mode 100644 python/packages/autogen-core/src/autogen_core/components/_type_prefix_subscription.py diff --git a/docs/design/04 - Agent and Topic ID Specs.md b/docs/design/04 - Agent and Topic ID Specs.md index ee872ab2ac1e..b0e0e0e94e60 100644 --- a/docs/design/04 - Agent and Topic ID Specs.md +++ b/docs/design/04 - Agent and Topic ID Specs.md @@ -34,7 +34,7 @@ This document describes the structure, constraints, and behavior of Agent IDs an - Type: `string` - Description: Topic type is usually defined by application code to mark the type of messages the topic is for. -- Constraints: UTF8 and only contain alphanumeric letters (a-z) and (0-9), or underscores (\_). A valid identifier cannot start with a number, or contain any spaces. +- Constraints: UTF8 and only contain alphanumeric letters (a-z) and (0-9), ':', '=', or underscores (\_). A valid identifier cannot start with a number, or contain any spaces. - Examples: - `GitHub_Issues` diff --git a/protos/agent_worker.proto b/protos/agent_worker.proto index 7b0b5245dd3e..61b00333cd24 100644 --- a/protos/agent_worker.proto +++ b/protos/agent_worker.proto @@ -63,9 +63,15 @@ message TypeSubscription { string agent_type = 2; } +message TypePrefixSubscription { + string topic_type_prefix = 1; + string agent_type = 2; +} + message Subscription { oneof subscription { TypeSubscription typeSubscription = 1; + TypePrefixSubscription typePrefixSubscription = 2; } } diff --git a/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py b/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py index 2c405710876a..f8ad4bbd14e9 100644 --- a/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py +++ b/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py @@ -47,7 +47,7 @@ ) from ..base._serialization import MessageSerializer, SerializationRegistry from ..base._type_helpers import ChannelArgumentType -from ..components import TypeSubscription +from ..components import TypePrefixSubscription, TypeSubscription from ._helpers import SubscriptionManager, get_impl from ._utils import GRPC_IMPORT_ERROR_STR from .protos import agent_worker_pb2, agent_worker_pb2_grpc @@ -705,27 +705,44 @@ async def try_get_underlying_agent_instance(self, id: AgentId, type: Type[T] = A async def add_subscription(self, subscription: Subscription) -> None: if self._host_connection is None: raise RuntimeError("Host connection is not set.") - if not isinstance(subscription, TypeSubscription): - raise ValueError("Only TypeSubscription is supported.") - # Add to local subscription manager. - await self._subscription_manager.add_subscription(subscription) # Create a future for the subscription response. future = asyncio.get_event_loop().create_future() request_id = await self._get_new_request_id() + + match subscription: + case TypeSubscription(topic_type=topic_type, agent_type=agent_type): + message = agent_worker_pb2.Message( + addSubscriptionRequest=agent_worker_pb2.AddSubscriptionRequest( + request_id=request_id, + subscription=agent_worker_pb2.Subscription( + typeSubscription=agent_worker_pb2.TypeSubscription( + topic_type=topic_type, agent_type=agent_type + ) + ), + ) + ) + case TypePrefixSubscription(topic_type_prefix=topic_type_prefix, agent_type=agent_type): + message = agent_worker_pb2.Message( + addSubscriptionRequest=agent_worker_pb2.AddSubscriptionRequest( + request_id=request_id, + subscription=agent_worker_pb2.Subscription( + typePrefixSubscription=agent_worker_pb2.TypePrefixSubscription( + topic_type_prefix=topic_type_prefix, agent_type=agent_type + ) + ), + ) + ) + case _: + raise ValueError("Unsupported subscription type.") + + # Add the future to the pending requests. self._pending_requests[request_id] = future + # Add to local subscription manager. + await self._subscription_manager.add_subscription(subscription) + # Send the subscription to the host. - message = agent_worker_pb2.Message( - addSubscriptionRequest=agent_worker_pb2.AddSubscriptionRequest( - request_id=request_id, - subscription=agent_worker_pb2.Subscription( - typeSubscription=agent_worker_pb2.TypeSubscription( - topic_type=subscription.topic_type, agent_type=subscription.agent_type - ) - ), - ) - ) await self._host_connection.send(message) # Wait for the subscription response. diff --git a/python/packages/autogen-core/src/autogen_core/application/_worker_runtime_host_servicer.py b/python/packages/autogen-core/src/autogen_core/application/_worker_runtime_host_servicer.py index 3da50c56f048..5cd2bf8ea9b8 100644 --- a/python/packages/autogen-core/src/autogen_core/application/_worker_runtime_host_servicer.py +++ b/python/packages/autogen-core/src/autogen_core/application/_worker_runtime_host_servicer.py @@ -4,7 +4,9 @@ from asyncio import Future, Task from typing import Any, Dict, Set -from ..base import TopicId +from autogen_core.components._type_prefix_subscription import TypePrefixSubscription + +from ..base import Subscription, TopicId from ..components import TypeSubscription from ._helpers import SubscriptionManager from ._utils import GRPC_IMPORT_ERROR_STR @@ -221,34 +223,46 @@ async def _process_add_subscription_request( self, add_subscription_req: agent_worker_pb2.AddSubscriptionRequest, client_id: int ) -> None: oneofcase = add_subscription_req.subscription.WhichOneof("subscription") + subscription: Subscription | None = None match oneofcase: case "typeSubscription": type_subscription_msg: agent_worker_pb2.TypeSubscription = ( add_subscription_req.subscription.typeSubscription ) - type_subscription = TypeSubscription( + subscription = TypeSubscription( topic_type=type_subscription_msg.topic_type, agent_type=type_subscription_msg.agent_type ) - try: - await self._subscription_manager.add_subscription(type_subscription) - subscription_ids = self._client_id_to_subscription_id_mapping.setdefault(client_id, set()) - subscription_ids.add(type_subscription.id) - success = True - error = None - except ValueError as e: - success = False - error = str(e) - # Send a response back to the client. - await self._send_queues[client_id].put( - agent_worker_pb2.Message( - addSubscriptionResponse=agent_worker_pb2.AddSubscriptionResponse( - request_id=add_subscription_req.request_id, success=success, error=error - ) - ) + + case "typePrefixSubscription": + type_prefix_subscription_msg: agent_worker_pb2.TypePrefixSubscription = ( + add_subscription_req.subscription.typePrefixSubscription + ) + subscription = TypePrefixSubscription( + topic_type_prefix=type_prefix_subscription_msg.topic_type_prefix, + agent_type=type_prefix_subscription_msg.agent_type, ) case None: logger.warning("Received empty subscription message") + if subscription is not None: + try: + await self._subscription_manager.add_subscription(subscription) + subscription_ids = self._client_id_to_subscription_id_mapping.setdefault(client_id, set()) + subscription_ids.add(subscription.id) + success = True + error = None + except ValueError as e: + success = False + error = str(e) + # Send a response back to the client. + await self._send_queues[client_id].put( + agent_worker_pb2.Message( + addSubscriptionResponse=agent_worker_pb2.AddSubscriptionResponse( + request_id=add_subscription_req.request_id, success=success, error=error + ) + ) + ) + async def GetState( # type: ignore self, request: agent_worker_pb2.AgentId, diff --git a/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.py b/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.py index 0637e866c4de..8f143d770aef 100644 --- a/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.py +++ b/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.py @@ -16,7 +16,7 @@ from google.protobuf import any_pb2 as google_dot_protobuf_dot_any__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x12\x61gent_worker.proto\x12\x06\x61gents\x1a\x10\x63loudevent.proto\x1a\x19google/protobuf/any.proto\"\'\n\x07TopicId\x12\x0c\n\x04type\x18\x01 \x01(\t\x12\x0e\n\x06source\x18\x02 \x01(\t\"$\n\x07\x41gentId\x12\x0c\n\x04type\x18\x01 \x01(\t\x12\x0b\n\x03key\x18\x02 \x01(\t\"E\n\x07Payload\x12\x11\n\tdata_type\x18\x01 \x01(\t\x12\x19\n\x11\x64\x61ta_content_type\x18\x02 \x01(\t\x12\x0c\n\x04\x64\x61ta\x18\x03 \x01(\x0c\"\x89\x02\n\nRpcRequest\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12$\n\x06source\x18\x02 \x01(\x0b\x32\x0f.agents.AgentIdH\x00\x88\x01\x01\x12\x1f\n\x06target\x18\x03 \x01(\x0b\x32\x0f.agents.AgentId\x12\x0e\n\x06method\x18\x04 \x01(\t\x12 \n\x07payload\x18\x05 \x01(\x0b\x32\x0f.agents.Payload\x12\x32\n\x08metadata\x18\x06 \x03(\x0b\x32 .agents.RpcRequest.MetadataEntry\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42\t\n\x07_source\"\xb8\x01\n\x0bRpcResponse\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12 \n\x07payload\x18\x02 \x01(\x0b\x32\x0f.agents.Payload\x12\r\n\x05\x65rror\x18\x03 \x01(\t\x12\x33\n\x08metadata\x18\x04 \x03(\x0b\x32!.agents.RpcResponse.MetadataEntry\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"\xe4\x01\n\x05\x45vent\x12\x12\n\ntopic_type\x18\x01 \x01(\t\x12\x14\n\x0ctopic_source\x18\x02 \x01(\t\x12$\n\x06source\x18\x03 \x01(\x0b\x32\x0f.agents.AgentIdH\x00\x88\x01\x01\x12 \n\x07payload\x18\x04 \x01(\x0b\x32\x0f.agents.Payload\x12-\n\x08metadata\x18\x05 \x03(\x0b\x32\x1b.agents.Event.MetadataEntry\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42\t\n\x07_source\"<\n\x18RegisterAgentTypeRequest\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x0c\n\x04type\x18\x02 \x01(\t\"^\n\x19RegisterAgentTypeResponse\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x12\n\x05\x65rror\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\":\n\x10TypeSubscription\x12\x12\n\ntopic_type\x18\x01 \x01(\t\x12\x12\n\nagent_type\x18\x02 \x01(\t\"T\n\x0cSubscription\x12\x34\n\x10typeSubscription\x18\x01 \x01(\x0b\x32\x18.agents.TypeSubscriptionH\x00\x42\x0e\n\x0csubscription\"X\n\x16\x41\x64\x64SubscriptionRequest\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12*\n\x0csubscription\x18\x02 \x01(\x0b\x32\x14.agents.Subscription\"\\\n\x17\x41\x64\x64SubscriptionResponse\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x12\n\x05\x65rror\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\"\x9d\x01\n\nAgentState\x12!\n\x08\x61gent_id\x18\x01 \x01(\x0b\x32\x0f.agents.AgentId\x12\x0c\n\x04\x65Tag\x18\x02 \x01(\t\x12\x15\n\x0b\x62inary_data\x18\x03 \x01(\x0cH\x00\x12\x13\n\ttext_data\x18\x04 \x01(\tH\x00\x12*\n\nproto_data\x18\x05 \x01(\x0b\x32\x14.google.protobuf.AnyH\x00\x42\x06\n\x04\x64\x61ta\"j\n\x10GetStateResponse\x12\'\n\x0b\x61gent_state\x18\x01 \x01(\x0b\x32\x12.agents.AgentState\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x12\n\x05\x65rror\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\"B\n\x11SaveStateResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x12\n\x05\x65rror\x18\x02 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\"\xc6\x03\n\x07Message\x12%\n\x07request\x18\x01 \x01(\x0b\x32\x12.agents.RpcRequestH\x00\x12\'\n\x08response\x18\x02 \x01(\x0b\x32\x13.agents.RpcResponseH\x00\x12\x1e\n\x05\x65vent\x18\x03 \x01(\x0b\x32\r.agents.EventH\x00\x12\x44\n\x18registerAgentTypeRequest\x18\x04 \x01(\x0b\x32 .agents.RegisterAgentTypeRequestH\x00\x12\x46\n\x19registerAgentTypeResponse\x18\x05 \x01(\x0b\x32!.agents.RegisterAgentTypeResponseH\x00\x12@\n\x16\x61\x64\x64SubscriptionRequest\x18\x06 \x01(\x0b\x32\x1e.agents.AddSubscriptionRequestH\x00\x12\x42\n\x17\x61\x64\x64SubscriptionResponse\x18\x07 \x01(\x0b\x32\x1f.agents.AddSubscriptionResponseH\x00\x12,\n\ncloudEvent\x18\x08 \x01(\x0b\x32\x16.cloudevent.CloudEventH\x00\x42\t\n\x07message2\xb2\x01\n\x08\x41gentRpc\x12\x33\n\x0bOpenChannel\x12\x0f.agents.Message\x1a\x0f.agents.Message(\x01\x30\x01\x12\x35\n\x08GetState\x12\x0f.agents.AgentId\x1a\x18.agents.GetStateResponse\x12:\n\tSaveState\x12\x12.agents.AgentState\x1a\x19.agents.SaveStateResponseB!\xaa\x02\x1eMicrosoft.AutoGen.Abstractionsb\x06proto3') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x12\x61gent_worker.proto\x12\x06\x61gents\x1a\x10\x63loudevent.proto\x1a\x19google/protobuf/any.proto\"\'\n\x07TopicId\x12\x0c\n\x04type\x18\x01 \x01(\t\x12\x0e\n\x06source\x18\x02 \x01(\t\"$\n\x07\x41gentId\x12\x0c\n\x04type\x18\x01 \x01(\t\x12\x0b\n\x03key\x18\x02 \x01(\t\"E\n\x07Payload\x12\x11\n\tdata_type\x18\x01 \x01(\t\x12\x19\n\x11\x64\x61ta_content_type\x18\x02 \x01(\t\x12\x0c\n\x04\x64\x61ta\x18\x03 \x01(\x0c\"\x89\x02\n\nRpcRequest\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12$\n\x06source\x18\x02 \x01(\x0b\x32\x0f.agents.AgentIdH\x00\x88\x01\x01\x12\x1f\n\x06target\x18\x03 \x01(\x0b\x32\x0f.agents.AgentId\x12\x0e\n\x06method\x18\x04 \x01(\t\x12 \n\x07payload\x18\x05 \x01(\x0b\x32\x0f.agents.Payload\x12\x32\n\x08metadata\x18\x06 \x03(\x0b\x32 .agents.RpcRequest.MetadataEntry\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42\t\n\x07_source\"\xb8\x01\n\x0bRpcResponse\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12 \n\x07payload\x18\x02 \x01(\x0b\x32\x0f.agents.Payload\x12\r\n\x05\x65rror\x18\x03 \x01(\t\x12\x33\n\x08metadata\x18\x04 \x03(\x0b\x32!.agents.RpcResponse.MetadataEntry\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"\xe4\x01\n\x05\x45vent\x12\x12\n\ntopic_type\x18\x01 \x01(\t\x12\x14\n\x0ctopic_source\x18\x02 \x01(\t\x12$\n\x06source\x18\x03 \x01(\x0b\x32\x0f.agents.AgentIdH\x00\x88\x01\x01\x12 \n\x07payload\x18\x04 \x01(\x0b\x32\x0f.agents.Payload\x12-\n\x08metadata\x18\x05 \x03(\x0b\x32\x1b.agents.Event.MetadataEntry\x1a/\n\rMetadataEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\x42\t\n\x07_source\"<\n\x18RegisterAgentTypeRequest\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x0c\n\x04type\x18\x02 \x01(\t\"^\n\x19RegisterAgentTypeResponse\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x12\n\x05\x65rror\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\":\n\x10TypeSubscription\x12\x12\n\ntopic_type\x18\x01 \x01(\t\x12\x12\n\nagent_type\x18\x02 \x01(\t\"G\n\x16TypePrefixSubscription\x12\x19\n\x11topic_type_prefix\x18\x01 \x01(\t\x12\x12\n\nagent_type\x18\x02 \x01(\t\"\x96\x01\n\x0cSubscription\x12\x34\n\x10typeSubscription\x18\x01 \x01(\x0b\x32\x18.agents.TypeSubscriptionH\x00\x12@\n\x16typePrefixSubscription\x18\x02 \x01(\x0b\x32\x1e.agents.TypePrefixSubscriptionH\x00\x42\x0e\n\x0csubscription\"X\n\x16\x41\x64\x64SubscriptionRequest\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12*\n\x0csubscription\x18\x02 \x01(\x0b\x32\x14.agents.Subscription\"\\\n\x17\x41\x64\x64SubscriptionResponse\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x12\n\x05\x65rror\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\"\x9d\x01\n\nAgentState\x12!\n\x08\x61gent_id\x18\x01 \x01(\x0b\x32\x0f.agents.AgentId\x12\x0c\n\x04\x65Tag\x18\x02 \x01(\t\x12\x15\n\x0b\x62inary_data\x18\x03 \x01(\x0cH\x00\x12\x13\n\ttext_data\x18\x04 \x01(\tH\x00\x12*\n\nproto_data\x18\x05 \x01(\x0b\x32\x14.google.protobuf.AnyH\x00\x42\x06\n\x04\x64\x61ta\"j\n\x10GetStateResponse\x12\'\n\x0b\x61gent_state\x18\x01 \x01(\x0b\x32\x12.agents.AgentState\x12\x0f\n\x07success\x18\x02 \x01(\x08\x12\x12\n\x05\x65rror\x18\x03 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\"B\n\x11SaveStateResponse\x12\x0f\n\x07success\x18\x01 \x01(\x08\x12\x12\n\x05\x65rror\x18\x02 \x01(\tH\x00\x88\x01\x01\x42\x08\n\x06_error\"\xc6\x03\n\x07Message\x12%\n\x07request\x18\x01 \x01(\x0b\x32\x12.agents.RpcRequestH\x00\x12\'\n\x08response\x18\x02 \x01(\x0b\x32\x13.agents.RpcResponseH\x00\x12\x1e\n\x05\x65vent\x18\x03 \x01(\x0b\x32\r.agents.EventH\x00\x12\x44\n\x18registerAgentTypeRequest\x18\x04 \x01(\x0b\x32 .agents.RegisterAgentTypeRequestH\x00\x12\x46\n\x19registerAgentTypeResponse\x18\x05 \x01(\x0b\x32!.agents.RegisterAgentTypeResponseH\x00\x12@\n\x16\x61\x64\x64SubscriptionRequest\x18\x06 \x01(\x0b\x32\x1e.agents.AddSubscriptionRequestH\x00\x12\x42\n\x17\x61\x64\x64SubscriptionResponse\x18\x07 \x01(\x0b\x32\x1f.agents.AddSubscriptionResponseH\x00\x12,\n\ncloudEvent\x18\x08 \x01(\x0b\x32\x16.cloudevent.CloudEventH\x00\x42\t\n\x07message2\xb2\x01\n\x08\x41gentRpc\x12\x33\n\x0bOpenChannel\x12\x0f.agents.Message\x1a\x0f.agents.Message(\x01\x30\x01\x12\x35\n\x08GetState\x12\x0f.agents.AgentId\x1a\x18.agents.GetStateResponse\x12:\n\tSaveState\x12\x12.agents.AgentState\x1a\x19.agents.SaveStateResponseB!\xaa\x02\x1eMicrosoft.AutoGen.Abstractionsb\x06proto3') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -54,20 +54,22 @@ _globals['_REGISTERAGENTTYPERESPONSE']._serialized_end=1067 _globals['_TYPESUBSCRIPTION']._serialized_start=1069 _globals['_TYPESUBSCRIPTION']._serialized_end=1127 - _globals['_SUBSCRIPTION']._serialized_start=1129 - _globals['_SUBSCRIPTION']._serialized_end=1213 - _globals['_ADDSUBSCRIPTIONREQUEST']._serialized_start=1215 - _globals['_ADDSUBSCRIPTIONREQUEST']._serialized_end=1303 - _globals['_ADDSUBSCRIPTIONRESPONSE']._serialized_start=1305 - _globals['_ADDSUBSCRIPTIONRESPONSE']._serialized_end=1397 - _globals['_AGENTSTATE']._serialized_start=1400 - _globals['_AGENTSTATE']._serialized_end=1557 - _globals['_GETSTATERESPONSE']._serialized_start=1559 - _globals['_GETSTATERESPONSE']._serialized_end=1665 - _globals['_SAVESTATERESPONSE']._serialized_start=1667 - _globals['_SAVESTATERESPONSE']._serialized_end=1733 - _globals['_MESSAGE']._serialized_start=1736 - _globals['_MESSAGE']._serialized_end=2190 - _globals['_AGENTRPC']._serialized_start=2193 - _globals['_AGENTRPC']._serialized_end=2371 + _globals['_TYPEPREFIXSUBSCRIPTION']._serialized_start=1129 + _globals['_TYPEPREFIXSUBSCRIPTION']._serialized_end=1200 + _globals['_SUBSCRIPTION']._serialized_start=1203 + _globals['_SUBSCRIPTION']._serialized_end=1353 + _globals['_ADDSUBSCRIPTIONREQUEST']._serialized_start=1355 + _globals['_ADDSUBSCRIPTIONREQUEST']._serialized_end=1443 + _globals['_ADDSUBSCRIPTIONRESPONSE']._serialized_start=1445 + _globals['_ADDSUBSCRIPTIONRESPONSE']._serialized_end=1537 + _globals['_AGENTSTATE']._serialized_start=1540 + _globals['_AGENTSTATE']._serialized_end=1697 + _globals['_GETSTATERESPONSE']._serialized_start=1699 + _globals['_GETSTATERESPONSE']._serialized_end=1805 + _globals['_SAVESTATERESPONSE']._serialized_start=1807 + _globals['_SAVESTATERESPONSE']._serialized_end=1873 + _globals['_MESSAGE']._serialized_start=1876 + _globals['_MESSAGE']._serialized_end=2330 + _globals['_AGENTRPC']._serialized_start=2333 + _globals['_AGENTRPC']._serialized_end=2511 # @@protoc_insertion_point(module_scope) diff --git a/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.pyi b/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.pyi index 522124ab8891..728bfafcc81a 100644 --- a/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.pyi +++ b/python/packages/autogen-core/src/autogen_core/application/protos/agent_worker_pb2.pyi @@ -273,21 +273,43 @@ class TypeSubscription(google.protobuf.message.Message): global___TypeSubscription = TypeSubscription +@typing.final +class TypePrefixSubscription(google.protobuf.message.Message): + DESCRIPTOR: google.protobuf.descriptor.Descriptor + + TOPIC_TYPE_PREFIX_FIELD_NUMBER: builtins.int + AGENT_TYPE_FIELD_NUMBER: builtins.int + topic_type_prefix: builtins.str + agent_type: builtins.str + def __init__( + self, + *, + topic_type_prefix: builtins.str = ..., + agent_type: builtins.str = ..., + ) -> None: ... + def ClearField(self, field_name: typing.Literal["agent_type", b"agent_type", "topic_type_prefix", b"topic_type_prefix"]) -> None: ... + +global___TypePrefixSubscription = TypePrefixSubscription + @typing.final class Subscription(google.protobuf.message.Message): DESCRIPTOR: google.protobuf.descriptor.Descriptor TYPESUBSCRIPTION_FIELD_NUMBER: builtins.int + TYPEPREFIXSUBSCRIPTION_FIELD_NUMBER: builtins.int @property def typeSubscription(self) -> global___TypeSubscription: ... + @property + def typePrefixSubscription(self) -> global___TypePrefixSubscription: ... def __init__( self, *, typeSubscription: global___TypeSubscription | None = ..., + typePrefixSubscription: global___TypePrefixSubscription | None = ..., ) -> None: ... - def HasField(self, field_name: typing.Literal["subscription", b"subscription", "typeSubscription", b"typeSubscription"]) -> builtins.bool: ... - def ClearField(self, field_name: typing.Literal["subscription", b"subscription", "typeSubscription", b"typeSubscription"]) -> None: ... - def WhichOneof(self, oneof_group: typing.Literal["subscription", b"subscription"]) -> typing.Literal["typeSubscription"] | None: ... + def HasField(self, field_name: typing.Literal["subscription", b"subscription", "typePrefixSubscription", b"typePrefixSubscription", "typeSubscription", b"typeSubscription"]) -> builtins.bool: ... + def ClearField(self, field_name: typing.Literal["subscription", b"subscription", "typePrefixSubscription", b"typePrefixSubscription", "typeSubscription", b"typeSubscription"]) -> None: ... + def WhichOneof(self, oneof_group: typing.Literal["subscription", b"subscription"]) -> typing.Literal["typeSubscription", "typePrefixSubscription"] | None: ... global___Subscription = Subscription diff --git a/python/packages/autogen-core/src/autogen_core/components/__init__.py b/python/packages/autogen-core/src/autogen_core/components/__init__.py index 9ad8bdb35a19..4c4d02f3be25 100644 --- a/python/packages/autogen-core/src/autogen_core/components/__init__.py +++ b/python/packages/autogen-core/src/autogen_core/components/__init__.py @@ -7,6 +7,7 @@ from ._default_topic import DefaultTopicId from ._image import Image from ._routed_agent import RoutedAgent, TypeRoutedAgent, event, message_handler, rpc +from ._type_prefix_subscription import TypePrefixSubscription from ._type_subscription import TypeSubscription from ._types import FunctionCall @@ -24,4 +25,5 @@ "DefaultTopicId", "default_subscription", "type_subscription", + "TypePrefixSubscription", ] diff --git a/python/packages/autogen-core/src/autogen_core/components/_type_prefix_subscription.py b/python/packages/autogen-core/src/autogen_core/components/_type_prefix_subscription.py new file mode 100644 index 000000000000..d71b587d1418 --- /dev/null +++ b/python/packages/autogen-core/src/autogen_core/components/_type_prefix_subscription.py @@ -0,0 +1,63 @@ +import uuid + +from ..base import AgentId, Subscription, TopicId +from ..base.exceptions import CantHandleException + + +class TypePrefixSubscription(Subscription): + """This subscription matches on topics based on a prefix of the type and maps to agents using the source of the topic as the agent key. + + This subscription causes each source to have its own agent instance. + + Example: + + .. code-block:: python + + from autogen_core.components import TypePrefixSubscription + + subscription = TypePrefixSubscription(topic_type_prefix="t1", agent_type="a1") + + In this case: + + - A topic_id with type `t1` and source `s1` will be handled by an agent of type `a1` with key `s1` + - A topic_id with type `t1` and source `s2` will be handled by an agent of type `a1` with key `s2`. + - A topic_id with type `t1SUFFIX` and source `s2` will be handled by an agent of type `a1` with key `s2`. + + Args: + topic_type_prefix (str): Topic type prefix to match against + agent_type (str): Agent type to handle this subscription + """ + + def __init__(self, topic_type_prefix: str, agent_type: str): + self._topic_type_prefix = topic_type_prefix + self._agent_type = agent_type + self._id = str(uuid.uuid4()) + + @property + def id(self) -> str: + return self._id + + @property + def topic_type_prefix(self) -> str: + return self._topic_type_prefix + + @property + def agent_type(self) -> str: + return self._agent_type + + def is_match(self, topic_id: TopicId) -> bool: + return topic_id.type.startswith(self._topic_type_prefix) + + def map_to_agent(self, topic_id: TopicId) -> AgentId: + if not self.is_match(topic_id): + raise CantHandleException("TopicId does not match the subscription") + + return AgentId(type=self._agent_type, key=topic_id.source) + + def __eq__(self, other: object) -> bool: + if not isinstance(other, TypePrefixSubscription): + return False + + return self.id == other.id or ( + self.agent_type == other.agent_type and self.topic_type_prefix == other.topic_type_prefix + ) diff --git a/python/packages/autogen-core/src/autogen_core/components/_type_subscription.py b/python/packages/autogen-core/src/autogen_core/components/_type_subscription.py index 92709a457aec..94def76595d5 100644 --- a/python/packages/autogen-core/src/autogen_core/components/_type_subscription.py +++ b/python/packages/autogen-core/src/autogen_core/components/_type_subscription.py @@ -1,7 +1,6 @@ import uuid -from typing import TypeVar -from ..base import AgentId, BaseAgent, Subscription, TopicId +from ..base import AgentId, Subscription, TopicId from ..base.exceptions import CantHandleException @@ -59,6 +58,3 @@ def __eq__(self, other: object) -> bool: return False return self.id == other.id or (self.agent_type == other.agent_type and self.topic_type == other.topic_type) - - -BaseAgentType = TypeVar("BaseAgentType", bound="BaseAgent") From cf80b1bc14c2376f33ae07bf57d4802077284727 Mon Sep 17 00:00:00 2001 From: Jack Gerrits Date: Tue, 26 Nov 2024 16:09:14 -0500 Subject: [PATCH 3/3] Add request_id parameter (#4384) --- .../application/_single_threaded_agent_runtime.py | 10 ++++++++++ .../src/autogen_core/application/_worker_runtime.py | 10 ++++++++++ .../src/autogen_core/base/_agent_runtime.py | 4 +++- .../src/autogen_core/base/_message_context.py | 1 + 4 files changed, 24 insertions(+), 1 deletion(-) diff --git a/python/packages/autogen-core/src/autogen_core/application/_single_threaded_agent_runtime.py b/python/packages/autogen-core/src/autogen_core/application/_single_threaded_agent_runtime.py index 52d24c64d0cb..3d81f15eb330 100644 --- a/python/packages/autogen-core/src/autogen_core/application/_single_threaded_agent_runtime.py +++ b/python/packages/autogen-core/src/autogen_core/application/_single_threaded_agent_runtime.py @@ -4,6 +4,7 @@ import inspect import logging import threading +import uuid import warnings from asyncio import CancelledError, Future, Task from collections.abc import Sequence @@ -53,6 +54,7 @@ class PublishMessageEnvelope: sender: AgentId | None topic_id: TopicId metadata: EnvelopeMetadata | None = None + message_id: str @dataclass(kw_only=True) @@ -256,6 +258,7 @@ async def publish_message( *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, + message_id: str | None = None, ) -> None: with self._tracer_helper.trace_block( "create", @@ -268,6 +271,9 @@ async def publish_message( content = message.__dict__ if hasattr(message, "__dict__") else message logger.info(f"Publishing message of type {type(message).__name__} to all subscribers: {content}") + if message_id is None: + message_id = str(uuid.uuid4()) + # event_logger.info( # MessageEvent( # payload=message, @@ -285,6 +291,7 @@ async def publish_message( sender=sender, topic_id=topic_id, metadata=get_telemetry_envelope_metadata(), + message_id=message_id, ) ) @@ -327,6 +334,8 @@ async def _process_send(self, message_envelope: SendMessageEnvelope) -> None: topic_id=None, is_rpc=True, cancellation_token=message_envelope.cancellation_token, + # Will be fixed when send API removed + message_id="NOT_DEFINED_TODO_FIX", ) with MessageHandlerContext.populate_context(recipient_agent.id): response = await recipient_agent.on_message( @@ -385,6 +394,7 @@ async def _process_publish(self, message_envelope: PublishMessageEnvelope) -> No topic_id=message_envelope.topic_id, is_rpc=False, cancellation_token=message_envelope.cancellation_token, + message_id=message_envelope.message_id, ) agent = await self._get_agent(agent_id) diff --git a/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py b/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py index f8ad4bbd14e9..0e5fb933a08e 100644 --- a/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py +++ b/python/packages/autogen-core/src/autogen_core/application/_worker_runtime.py @@ -3,6 +3,7 @@ import json import logging import signal +import uuid import warnings from asyncio import Future, Task from collections import defaultdict @@ -371,11 +372,17 @@ async def publish_message( *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, + message_id: str | None = None, ) -> None: if not self._running: raise ValueError("Runtime must be running when publishing message.") if self._host_connection is None: raise RuntimeError("Host connection is not set.") + if message_id is None: + message_id = str(uuid.uuid4()) + + # TODO: consume message_id + message_type = self._serialization_registry.type_name(message) with self._trace_helper.trace_block( "create", topic_id, parent=None, extraAttributes={"message_type": message_type} @@ -447,6 +454,7 @@ async def _process_request(self, request: agent_worker_pb2.RpcRequest) -> None: topic_id=None, is_rpc=True, cancellation_token=CancellationToken(), + message_id=request.request_id, ) # Call the receiving agent. @@ -530,11 +538,13 @@ async def _process_event(self, event: agent_worker_pb2.Event) -> None: for agent_id in recipients: if agent_id == sender: continue + # TODO: consume message_id message_context = MessageContext( sender=sender, topic_id=topic_id, is_rpc=False, cancellation_token=CancellationToken(), + message_id="NOT_DEFINED_TODO_FIX", ) agent = await self._get_agent(agent_id) with MessageHandlerContext.populate_context(agent.id): diff --git a/python/packages/autogen-core/src/autogen_core/base/_agent_runtime.py b/python/packages/autogen-core/src/autogen_core/base/_agent_runtime.py index a8e7f0096324..27c37ad9f349 100644 --- a/python/packages/autogen-core/src/autogen_core/base/_agent_runtime.py +++ b/python/packages/autogen-core/src/autogen_core/base/_agent_runtime.py @@ -55,6 +55,7 @@ async def publish_message( *, sender: AgentId | None = None, cancellation_token: CancellationToken | None = None, + message_id: str | None = None, ) -> None: """Publish a message to all agents in the given namespace, or if no namespace is provided, the namespace of the sender. @@ -64,7 +65,8 @@ async def publish_message( message (Any): The message to publish. topic (TopicId): The topic to publish the message to. sender (AgentId | None, optional): The agent which sent the message. Defaults to None. - cancellation_token (CancellationToken | None, optional): Token used to cancel an in progress . Defaults to None. + cancellation_token (CancellationToken | None, optional): Token used to cancel an in progress. Defaults to None. + message_id (str | None, optional): The message id. If None, a new message id will be generated. Defaults to None. This message id must be unique. and is recommended to be a UUID. Raises: UndeliverableException: If the message cannot be delivered. diff --git a/python/packages/autogen-core/src/autogen_core/base/_message_context.py b/python/packages/autogen-core/src/autogen_core/base/_message_context.py index 0a2c2973bc01..c5c00559ed0e 100644 --- a/python/packages/autogen-core/src/autogen_core/base/_message_context.py +++ b/python/packages/autogen-core/src/autogen_core/base/_message_context.py @@ -11,3 +11,4 @@ class MessageContext: topic_id: TopicId | None is_rpc: bool cancellation_token: CancellationToken + message_id: str