系列第7篇：Python+Go构建企业级AI Agent实战指南（7/13）

标签： Go | Python | gRPC | RabbitMQ | 通信机制

---

一、开篇：双栈通信的核心挑战

Python负责AI推理，Go负责基础设施——这是2026年的主流架构。但两者如何高效通信？

核心挑战：

1. 性能：Python GIL限制，如何充分利用多核？
2. 可靠：任务执行超时、失败如何重试？
3. 扩展：水平扩展时如何负载均衡？

本文将深入三种通信方案：HTTP REST、gRPC、消息队列。

---

二、方案对比

方案	延迟	吞吐量	复杂度	适用场景
HTTP REST	10-50ms	中等	低	简单同步调用
gRPC	1-5ms	高	中	高性能服务间通信
消息队列	异步	极高	高	高可靠、削峰填谷

---

三、gRPC实战

3.1 定义Proto

// proto/agent.proto
syntax = "proto3";

package agent;

service AgentService {
  rpc RunTask (TaskRequest) returns (TaskResponse);
  rpc StreamTask (TaskRequest) returns (stream TaskProgress);
  rpc HealthCheck (HealthRequest) returns (HealthResponse);
}

message TaskRequest {
  string task_id = 1;
  string agent_type = 2;
  string payload = 3;  // JSON string
  int32 timeout_seconds = 4;
}

message TaskResponse {
  bool success = 1;
  string result = 2;  // JSON string
  string error = 3;
  int64 execution_time_ms = 4;
}

message TaskProgress {
  string task_id = 1;
  string stage = 2;
  int32 progress_percent = 3;
  string message = 4;
}

message HealthRequest {}

message HealthResponse {
  bool healthy = 1;
  string version = 2;
}

3.2 Go服务端

// internal/grpc/server.go
package grpc

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"google.golang.org/grpc"
	pb "agent-gateway/proto"
)

type AgentServer struct {
	pb.UnimplementedAgentServiceServer
	pythonClient *service.PythonClient
}

func NewAgentServer(pythonAddr string) *AgentServer {
	return &AgentServer{
		pythonClient: service.NewPythonClient(pythonAddr),
	}
}

func (s *AgentServer) RunTask(ctx context.Context, req *pb.TaskRequest) (*pb.TaskResponse, error) {
	start := time.Now()
	
	// 解析payload
	var input map[string]interface{}
	if err := json.Unmarshal([]byte(req.Payload), &input); err != nil {
		return &pb.TaskResponse{
			Success: false,
			Error:   fmt.Sprintf("invalid payload: %v", err),
		}, nil
	}
	
	// 调用Python服务
	agentReq := &service.AgentRequest{
		AgentType: req.AgentType,
		Input:     input,
	}
	
	// 设置超时
	if req.TimeoutSeconds > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, time.Duration(req.TimeoutSeconds)*time.Second)
		defer cancel()
	}
	
	resp, err := s.pythonClient.RunAgent(ctx, agentReq)
	executionTime := time.Since(start).Milliseconds()
	
	if err != nil {
		return &pb.TaskResponse{
			Success:         false,
			Error:           err.Error(),
			ExecutionTimeMs: executionTime,
		}, nil
	}
	
	resultJSON, _ := json.Marshal(resp.Result)
	
	return &pb.TaskResponse{
		Success:         resp.Success,
		Result:          string(resultJSON),
		Error:           resp.Error,
		ExecutionTimeMs: executionTime,
	}, nil
}

func (s *AgentServer) StreamTask(req *pb.TaskRequest, stream pb.AgentService_StreamTaskServer) error {
	// 模拟流式进度
	stages := []struct {
		stage   string
		message string
		percent int32
	}{
		{"init", "初始化任务", 10},
		{"load", "加载模型", 30},
		{"process", "处理数据", 60},
		{"analyze", "分析结果", 90},
		{"complete", "任务完成", 100},
	}
	
	for _, stage := range stages {
		progress := &pb.TaskProgress{
			TaskId:          req.TaskId,
			Stage:           stage.stage,
			ProgressPercent: stage.percent,
			Message:         stage.message,
		}
		
		if err := stream.Send(progress); err != nil {
			return err
		}
		
		time.Sleep(500 * time.Millisecond)
	}
	
	return nil
}

func (s *AgentServer) HealthCheck(ctx context.Context, req *pb.HealthRequest) (*pb.HealthResponse, error) {
	return &pb.HealthResponse{
		Healthy: true,
		Version: "1.0.0",
	}, nil
}

3.3 Python gRPC客户端

# python_agent/grpc_server.py
from concurrent import futures
import grpc
import json
import time

import agent_pb2
import agent_pb2_grpc

class AgentServiceServicer(agent_pb2_grpc.AgentServiceServicer):
    """Agent服务实现"""
    
    def RunTask(self, request, context):
        """执行任务"""
        print(f"🚀 收到任务: {request.task_id}")
        
        try:
            # 解析输入
            input_data = json.loads(request.payload)
            
            # 模拟Agent执行
            result = self.execute_agent(
                agent_type=request.agent_type,
                input_data=input_data
            )
            
            return agent_pb2.TaskResponse(
                success=True,
                result=json.dumps(result),
                execution_time_ms=int(time.time() * 1000)
            )
            
        except Exception as e:
            return agent_pb2.TaskResponse(
                success=False,
                error=str(e)
            )
    
    def StreamTask(self, request, context):
        """流式任务进度"""
        stages = [
            ("init", "初始化", 10),
            ("load", "加载模型", 30),
            ("process", "处理中", 60),
            ("analyze", "分析结果", 90),
            ("complete", "完成", 100),
        ]
        
        for stage, message, percent in stages:
            yield agent_pb2.TaskProgress(
                task_id=request.task_id,
                stage=stage,
                progress_percent=percent,
                message=message
            )
            time.sleep(0.5)
    
    def HealthCheck(self, request, context):
        """健康检查"""
        return agent_pb2.HealthResponse(
            healthy=True,
            version="1.0.0"
        )
    
    def execute_agent(self, agent_type, input_data):
        """执行Agent逻辑"""
        # 这里集成实际的Agent执行逻辑
        return {
            "agent_type": agent_type,
            "input": input_data,
            "output": f"Processed by {agent_type}",
            "timestamp": time.time()
        }

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    agent_pb2_grpc.add_AgentServiceServicer_to_server(
        AgentServiceServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    print("🚀 Python gRPC Server started on :50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

---

四、消息队列方案：RabbitMQ

4.1 Go生产者

// internal/mq/producer.go
package mq

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/streadway/amqp"
)

type Producer struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	queue   amqp.Queue
}

func NewProducer(amqpURL string) (*Producer, error) {
	conn, err := amqp.Dial(amqpURL)
	if err != nil {
		return nil, err
	}
	
	ch, err := conn.Channel()
	if err != nil {
		return nil, err
	}
	
	q, err := ch.QueueDeclare(
		"agent_tasks", // name
		true,          // durable
		false,         // delete when unused
		false,         // exclusive
		false,         // no-wait
		nil,           // arguments
	)
	if err != nil {
		return nil, err
	}
	
	return &Producer{
		conn:    conn,
		channel: ch,
		queue:   q,
	}, nil
}

func (p *Producer) PublishTask(ctx context.Context, taskID string, payload map[string]interface{}) error {
	body, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	
	return p.channel.Publish(
		"",           // exchange
		p.queue.Name, // routing key
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         body,
			MessageId:    taskID,
			DeliveryMode: amqp.Persistent,
		},
	)
}

4.2 Python消费者

# python_agent/mq_consumer.py
import pika
import json
import threading

class MQConsumer:
    """RabbitMQ消费者"""
    
    def __init__(self, amqp_url='amqp://guest:guest@localhost:5672/'):
        self.amqp_url = amqp_url
        self.connection = None
        self.channel = None
    
    def connect(self):
        """建立连接"""
        params = pika.URLParameters(self.amqp_url)
        self.connection = pika.BlockingConnection(params)
        self.channel = self.connection.channel()
        
        # 声明队列
        self.channel.queue_declare(queue='agent_tasks', durable=True)
        
        # 设置QoS，每次只取一个任务
        self.channel.basic_qos(prefetch_count=1)
    
    def process_task(self, body):
        """处理任务"""
        try:
            task = json.loads(body)
            print(f"🚀 处理任务: {task.get('task_id')}")
            
            # 执行Agent逻辑
            result = self.execute_agent(task)
            
            # 发送结果到结果队列
            self.send_result(task['task_id'], result)
            
            return True
        except Exception as e:
            print(f"❌ 任务处理失败: {e}")
            return False
    
    def execute_agent(self, task):
        """执行Agent"""
        agent_type = task.get('agent_type')
        input_data = task.get('input', {})
        
        # 这里调用实际的Agent
        return {
            'success': True,
            'agent_type': agent_type,
            'result': f'Processed: {input_data}'
        }
    
    def send_result(self, task_id, result):
        """发送结果"""
        self.channel.basic_publish(
            exchange='',
            routing_key='agent_results',
            body=json.dumps({
                'task_id': task_id,
                'result': result
            })
        )
    
    def callback(self, ch, method, properties, body):
        """消息回调"""
        success = self.process_task(body)
        
        if success:
            ch.basic_ack(delivery_tag=method.delivery_tag)
        else:
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    
    def start(self):
        """启动消费者"""
        self.connect()
        
        self.channel.basic_consume(
            queue='agent_tasks',
            on_message_callback=self.callback
        )
        
        print('🐰 MQ Consumer started, waiting for tasks...')
        self.channel.start_consuming()

if __name__ == '__main__':
    consumer = MQConsumer()
    consumer.start()

---

五、性能对比实测

测试环境：8核16G，本地网络

HTTP REST (1000并发):
  平均延迟: 45ms
  P99延迟: 120ms
  QPS: 2,200

gRPC (1000并发):
  平均延迟: 8ms
  P99延迟: 25ms
  QPS: 12,500

RabbitMQ (生产者-消费者):
  吞吐量: 15,000 msg/s
  延迟: 异步，不阻塞

---

六、选型建议

场景	推荐方案
简单同步调用	HTTP REST
高性能服务间通信	gRPC
削峰填谷、高可靠	RabbitMQ
流式输出	gRPC Streaming
混合架构	gRPC + MQ组合

---

系列文章导航： ← 6. Go语言实现AI Agent任务调度网关 7. Go与Python Agent通信机制（本文） 8. 双栈协同：Python+Go混合架构实战 →

---

本文首发于CSDN，转载请注明出处。

标签： Go | Python | gRPC | RabbitMQ | 通信机制