AutoGen 跨语言 Agent 互操作实践：Python 与 .NET 通过 gRPC 运行时互通的 Hello 示例

本文围绕 AutoGen 仓库中的跨语言（xlang）互操作示例 core_xlang_hello_python_agent 展开，讲解如何用 Python 编写的 Agent 通过 gRPC 工作区运行时连接 .NET Agent Host，与 .NET 端 Agent 基于 Protobuf 消息契约互通。读完本文，你将掌握该示例的完整运行方式（基于 .NET Aspire 编排）、Python 侧 GrpcWorkerAgentRuntime 的连接/订阅/发布机制，以及 Python 与 .NET 两端从消息到话题（Topic）路由的完整调用链。

一、示例定位：一个 Python Agent 与一个 .NET Agent 对话

示例 README 给出的目标非常聚焦：创建一个 Python Agent，让它与一个 .NET Agent 交互。其工作原理是：Python Agent 向 .NET 运行时（Agent Host / gRPC 网关）发送消息，由 .NET 运行时把消息中转（relay）给 .NET Agent，从而实现两种语言栈内 Agent 的互通。

这个示例所在的目录是 AutoGen 的集成测试资源之一（Microsoft.AutoGen.Integration.Tests.AppHosts），但它的实际运行入口在 dotnet/samples/Hello 示例的 Aspire App Host 中——.NET 端与 Python 端各自作为独立进程，通过同一个后端网关连接，构成最小可运行的跨语言 Agent 拓扑。

二、运行方式：用 .NET Aspire 一键拉起三个项目

按照 README 的原始步骤：

1. 检出 autogen 仓库后，进入 autogen/dotnet/samples/Hello/Hello.AppHost；
2. 执行 dotnet run 启动 .NET Aspire App Host。它会编排并启动三个项目：
   • Backend：.NET Agent Runtime（即 gRPC 网关后端）；
   • HelloAgent：.NET 端的 Agent 进程；
   • 本 Python Agent：hello_python_agent.py；
3. App Host 启动后会在浏览器中打开 Aspire Dashboard（README 给出的地址为 https://localhost:15887），可在此查看各组件的遥测与日志。

更完整的编排细节可以在 Hello.AppHost/Program.cs 中确认，其中关键代码为：

var backend = builder.AddProject<Projects.Microsoft_AutoGen_AgentHost>("backend")
    .WithExternalHttpEndpoints();
var client = builder.AddProject<Projects.HelloAgent>("HelloAgentsDotNET")
    .WithReference(backend)
    .WithEnvironment("AGENT_HOST", backend.GetEndpoint("https"))
    .WithEnvironment("STAY_ALIVE_ON_GOODBYE", "true")
    .WaitFor(backend);
// xlang 目前走 http —— 生产环境中应在容器间启用 TLS
builder.AddPythonApp("HelloAgentsPython",
        "../../../../python/samples/core_xlang_hello_python_agent",
        "hello_python_agent.py", "../../.venv")
    .WithReference(backend)
    .WithEnvironment("AGENT_HOST", backend.GetEndpoint("http"))
    .WithEnvironment("STAY_ALIVE_ON_GOODBYE", "true")
    .WithEnvironment("GRPC_DNS_RESOLVER", "native")
    .WithOtlpExporter()
    .WaitFor(client);

从这段编排代码可以看出三个关键环境约定：

环境变量	赋值来源	作用
AGENT_HOST	.NET 端取后端 https 端点；Python 端取 http 端点	两端 gRPC Worker 运行时连接的后端地址（Python 侧默认值回退为 http://localhost:50673）
STAY_ALIVE_ON_GOODBYE	true	让 .NET HelloAgent 收到 ConversationClosed 后不发布 Shutdown、不退出进程，保证演示会话持续可用
GRPC_DNS_RESOLVER	native	规避 gRPC Python 客户端在容器网络下 DNS 解析的问题

三、示例文件结构

该目录下与主题直接相关的文件如下：

文件	说明
hello_python_agent.py	Python Agent 主程序：连接运行时、注册 Agent、订阅话题、发布跨语言消息
user_input.py	UserProxy：让人以“Agent 身份”参与对话的 RoutedAgent
protos/	由 agent_events.proto 生成的 Python Protobuf/gRPC 桩代码（agent_events_pb2.py 等）
README.md	本文的关联文档，描述运行步骤与交互原理

其中 Python 端的消息契约与 .NET 端共用同一份 proto 定义 dotnet/samples/Hello/protos/agent_events.proto（包名 HelloAgents，.NET 命名空间映射为 Microsoft.AutoGen.Contracts），本目录下的 protos/ 是它生成到 Python 侧的产物。核心消息类型包括：

message Input { string message = 1; }
message Output { string message = 1; }
message NewMessageReceived { string message = 1; }
message ConversationClosed {
  string user_id = 1;
  string user_message = 2;
}
message Shutdown { string message = 1; }

这一“同一份 proto、两端各自生成桩”的做法，正是 AutoGen 跨语言互操作的消息层基础：Python 与 .NET 传递的是结构化 Protobuf 负载，而不是各自私有的对象模型。

四、Python Agent 主程序逐段解析

hellp_python_agent.py 的 main() 可以分为四个阶段，以下逐段对照源码讲解。

4.1 解析 AGENT_HOST 并创建 gRPC Worker 运行时

agentHost = os.getenv("AGENT_HOST") or "http://localhost:50673"
# grpc python bug - 只能使用主机名、不能带协议前缀：
if agentHost.startswith("http://"):
    agentHost = agentHost[7:]
if agentHost.startswith("https://"):
    agentHost = agentHost[8:]
runtime = GrpcWorkerAgentRuntime(
    host_address=agentHost,
    payload_serialization_format=PROTOBUF_DATA_CONTENT_TYPE,
)

三个要点：

• GrpcWorkerAgentRuntime 来自 autogen_ext.runtimes.grpc，实现类定义在 autogen-ext 的 _worker_runtime.py 中，它是 Python 侧以“工作区（worker）”身份接入分布式 Agent 运行时（.NET 端 Agent Host）的载体；
• payload_serialization_format=PROTOBUF_DATA_CONTENT_TYPE 声明消息负载使用 Protobuf 序列化——这与两端共享 proto 契约的设计一一对应；
• 源码注释明确指出了一个实际限制：gRPC Python 客户端只接受主机名，不能带 http:///https:// 前缀，因此这里手动剥离前缀。Aspire 注入的 AGENT_HOST 是带端口的完整 URL，这一步是必要的。

4.2 注册消息序列化器

await runtime.start()
runtime.add_message_serializer(try_get_known_serializers_for_type(NewMessageReceived))

try_get_known_serializers_for_type（来自 autogen_core）会为指定的 Protobuf 消息类型注册对应的序列化器，使运行时知道如何把 NewMessageReceived 这类消息编码/解码为线上字节。示例还针对 Output 类型同样注册了序列化器（见 4.4）。

4.3 注册 Agent 并建立订阅

await UserProxy.register(runtime, "HelloAgent", lambda: UserProxy())
await runtime.add_subscription(DefaultSubscription(agent_type="HelloAgent"))
await runtime.add_subscription(TypeSubscription(topic_type="HelloTopic", agent_type="HelloAgent"))
await runtime.add_subscription(TypeSubscription(topic_type="agents.NewMessageReceived", agent_type="HelloAgent"))
await runtime.add_subscription(TypeSubscription(topic_type="agents.ConversationClosed", agent_type="HelloAgent"))
await runtime.add_subscription(TypeSubscription(topic_type="agents.Output", agent_type="HelloAgent"))

这里体现了 AutoGen core 的两类路由原语：

• DefaultSubscription(agent_type=...)：订阅默认话题上发给指定 Agent 类型的消息，即“点对点”路由；
• TypeSubscription(topic_type=..., agent_type=...)：按“话题类型 × Agent 类型”的组合订阅，HelloTopic 对应 .NET 端发布对话消息的话题，而 agents.NewMessageReceived / agents.ConversationClosed / agents.Output 对应框架级 Agent 话题上的具体消息类型。

值得注意的是，Python 侧把 UserProxy 注册为名为 "HelloAgent" 的 Agent 类型——这正是它能在 .NET 端把消息路由到“HelloAgent”这一类型时接收到转发的原因（README 中“发送消息到 .NET 运行时、由其转发给 .NET Agent”的交互，订阅层面就是这样落地的）。

4.4 发布两条跨语言消息并挂起等待信号

new_message = NewMessageReceived(message="Hello from Python!")
await runtime.publish_message(
    message=new_message,
    topic_id=DefaultTopicId("HelloTopic", "HelloAgents/python"),
    sender=AgentId("HelloAgents", "python"),
)
output_message = Output(message="^v^v^v---Wild Hello from Python!---^v^v^v")
runtime.add_message_serializer(try_get_known_serializers_for_type(Output))
await runtime.publish_message(
    message=output_message,
    topic_id=DefaultTopicId("HelloTopic", "HelloAgents/python"),
    sender=AgentId("HelloAgents", "python"),
)
await runtime.stop_when_signal()

• 消息发布到 DefaultTopicId("HelloTopic", "HelloAgents/python")，发送者身份为 AgentId("HelloAgents", "python")，即一个跨语言身份标识（类型 + 路由）；
• HelloTopic 上的 NewMessageReceived 会被 .NET 端 HelloAgent 处理（见第六节）；
• 最后 runtime.stop_when_signal() 让进程在收到系统信号前持续运行，保持长连接会话（源码中被注释掉的 stop_when_idle() 是空闲即退出的替代策略）。

五、UserProxy：让人以 Agent 身份参与跨语言对话

user_input.py 定义了一个典型的 AutoGen core RoutedAgent：

class UserProxy(RoutedAgent):
    """An agent that allows the user to play the role of an agent in the conversation via input."""

    @message_handler
    async def handle_user_chat_input(self, message: input_types, ctx: MessageContext) -> None:
        if isinstance(message, Input):
            response = await self.ainput("User input ('exit' to quit): ")
            await self.publish_message(NewMessageReceived(message=response),
                                        topic_id=DefaultTopicId())
        elif isinstance(message, Output):
            logger.info(message.message)

• @message_handler 装饰的方法统一处理三种输入类型 Union[ConversationClosed, Input, Output]；
• 收到 Input 时通过 asyncio.to_thread(input, ...) 在不阻塞事件循环的前提下读取终端输入，并把用户输入包装成 NewMessageReceived 发布到默认话题——这条消息同样会经由 .NET 运行时到达 .NET Agent，即“人类 ↔ Python Agent ↔ .NET 运行时 ↔ .NET Agent”的人机链路；
• 收到 Output 时仅打印内容，作为跨语言通道的输出展示点。

六、.NET 端：消息如何被 HelloAgent 消费

6.1 HelloAgent 的处理器

HelloAgent.cs 是 .NET 端接收方，它声明了 [TypeSubscription("HelloTopic")]，并实现三个处理器：

public async ValueTask HandleAsync(NewMessageReceived item, MessageContext messageContext)
{
    Console.Out.WriteLine(item.Message);
    await this.PublishMessageAsync(
        new ConversationClosed { UserId = this.Id.Type, UserMessage = "Goodbye" },
        new TopicId("HelloTopic"));
}

处理链为：NewMessageReceived（Python 发来）→ 打印内容并发布 ConversationClosed → ConversationClosed 处理器打印“xxx said Goodbye”，若 STAY_ALIVE_ON_GOODBYE 不为 true 则发布 Shutdown → Shutdown 处理器调用 IHostApplicationLifetime.StopApplication() 停机。由于 App Host 显式注入了 STAY_ALIVE_ON_GOODBYE=true，会话在演示期间不会被关闭。

6.2 端点的启动方式

HelloAgent/Program.cs 展示了同一套 Agent 代码的两种部署形态：当存在 AGENT_HOST（或 --host 参数）时使用 appBuilder.AddGrpcAgentWorker(agentHost) 以 gRPC Worker 身份接入分布式后端；否则回退到 UseInProcessRuntime 的进程内运行时。本跨语言示例走的是前者：

appBuilder.AddGrpcAgentWorker(agentHost).AddAgent<HelloAgent>("HelloAgent");

启动后（sendHello 默认为真）它还会发布一条 "Hello World!" 的 NewMessageReceived 到 HelloTopic，与 Python 端发来的 "Hello from Python!" 汇入同一话题。

七、完整消息流时间线

综合两端源码，一次典型交互的流转如下：

1. Aspire App Host 启动 Backend（Agent Host）→ .NET HelloAgent（AGENT_HOST 指向 https 端点）→ Python hello_python_agent.py（AGENT_HOST 指向 http 端点，WaitFor 保证先后顺序）；
2. Python 端注册序列化器与订阅后，发布 NewMessageReceived("Hello from Python!") 到 HelloTopic，发送者 AgentId("HelloAgents", "python")；
3. .NET 运行时按话题路由把消息投递给订阅了 HelloTopic 的 HelloAgent，后者打印并回复 ConversationClosed 到 HelloTopic；
4. Python 端通过 TypeSubscription(topic_type="agents.ConversationClosed", agent_type="HelloAgent") 收到该消息，UserProxy 打印输出（Output 消息同理经 agents.Output 话题回传展示）；
5. 若配置了 Input，人类在终端输入的内容会包装成 NewMessageReceived 再次进入同一条跨语言链路。

这正对应 README 的结论：Python Agent 通过向 .NET 运行时发消息、由运行时中转给 .NET Agent 来完成交互。

八、集成测试视角：同一 Python Agent 的复用

从源码结构看，该目录之所以位于 Microsoft.AutoGen.Integration.Tests.AppHosts 下，是因为它同时被集成测试复用。XLangTests.AppHost/Program.cs 直接以相对路径 ../core_xlang_hello_python_agent 和脚本 hello_python_agent.py 拉起同一个 Python Agent，并支持用环境变量 XLANG_TEST_NO_DOTNET / XLANG_TEST_NO_PYTHON 分别裁剪掉 .NET 或 Python 侧，验证“单边故障不影响对端连接”这类集成行为。对应的上层集成测试位于 Microsoft.AutoGen.Integration.Tests，例如 HelloAppHostIntegrationTests.cs 即针对 Hello 系列 App Host 的端到端验证。

九、要点小结与适用前提

• 消息契约先行：跨语言互通的第一步是共享 agent_events.proto 并生成双端桩代码，Python 侧的 try_get_known_serializers_for_type 与 .NET 侧的 Microsoft.AutoGen.Contracts 命名空间映射（option csharp_namespace）分别承担序列化与类型绑定；
• 运行时选型：Python 侧使用 GrpcWorkerAgentRuntime（autogen_ext.runtimes.grpc）并以 PROTOBUF_DATA_CONTENT_TYPE 承载负载；.NET 侧使用 AddGrpcAgentWorker 接入同一 Agent Host；
• 订阅决定可达性：DefaultSubscription 与 TypeSubscription 的组合（默认话题、业务话题 HelloTopic、框架话题 agents.*）是消息能否跨语言路由到目标 Agent 的前提；
• 适用前提与限制：该示例面向 .NET Aspire 本地开发环境运行（dotnet run 于 Hello.AppHost），Aspire 的 Python 应用资源在源码注释中标注为评估期 API；示例中 xlang 通道当前走 http（注释建议生产环境在容器间启用 TLS）；AGENT_HOST 在 Python 侧必须去掉协议前缀才能被 gRPC Python 客户端接受。

掌握这套“同一 proto 契约 + gRPC Worker 运行时 + 话题/Agent 订阅”的三要素组合后，即可把本示例中 Hello 级别的交互扩展为任意 Python/.NET 混合的多 Agent 系统。