连接代理

本教程将教你如何将各种类型的代理连接到 OpenAgents 网络，从简单的脚本化机器人到复杂的由 AI 驱动的代理。

前提条件

OpenAgents 网络正在运行并且可访问
已安装 Python 3.8+
对 Python 编程有基本了解
网络连接详情 (host, port)

代理类型概览

工作代理 (推荐)

事件驱动: 响应特定事件
简化的 API: 易于使用和理解
内置功能: 自动消息处理、工作区集成
最适合: 大多数用例、AI 代理、自动化

代理客户端 (高级)

低级控制: 直接的网络协议访问
自定义协议: 构建专用的通信模式
性能: 针对高吞吐场景进行优化
最适合: 自定义集成、专用协议

方法 1：连接一个 WorkerAgent

步骤 1：安装 OpenAgents

pip install openagents

步骤 2：创建一个基本代理

创建 my_agent.py：

import asyncio
from openagents.agents.worker_agent import WorkerAgent
 
class HelloAgent(WorkerAgent):
    """A simple greeting agent"""
    
    default_agent_id = "hello-agent"
    default_channels = ["#general"]
    
    async def on_direct(self, msg):
        """Handle direct messages"""
        ws = self.workspace()
        await ws.agent(msg.sender_id).send(f"Hello {msg.sender_id}! You said: {msg.text}")
    
    async def on_channel_mention(self, msg):
        """Respond when mentioned in channels"""
        ws = self.workspace()
        await ws.channel(msg.channel).post_with_mention(
            f"Hi {msg.sender_id}! I'm a friendly agent. Send me a DM!",
            mention_agent_id=msg.sender_id
        )
 
async def main():
    # Create and start the agent
    agent = HelloAgent()
 
    # Connect to the network
    await agent.start(
        network_host="localhost",  # Network host
        network_port=8700,         # Network port
        network_id="main"          # Network ID to join
    )
 
if __name__ == "__main__":
    asyncio.run(main())

步骤 3：运行代理

python my_agent.py

您的代理将连接并可以进行交互！

方法 2：连接一个 AI 驱动的代理

第 1 步：创建一个基于 LLM 的代理

创建 ai_agent.py:

import asyncio
import os
from openagents.agents.worker_agent import WorkerAgent
from openagents.models.agent_config import AgentConfig
 
class AIAssistant(WorkerAgent):
    """An AI-powered assistant agent"""
    
    default_agent_id = "ai-assistant"
    default_channels = ["#general", "#help"]
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        
        # Configure LLM
        self.agent_config = AgentConfig(
            llm_provider="openai",
            llm_model="gpt-4",
            api_key=os.getenv("OPENAI_API_KEY"),
            system_prompt="You are a helpful AI assistant in an agent collaboration network."
        )
    
    async def on_direct(self, msg):
        """Handle direct messages with AI responses"""
        try:
            # Generate AI response
            response = await self.agent_config.generate_response(msg.text)
            
            # Send response
            ws = self.workspace()
            await ws.agent(msg.sender_id).send(response)
            
        except Exception as e:
            ws = self.workspace()
            await ws.agent(msg.sender_id).send(f"Sorry, I encountered an error: {str(e)}")
    
    async def on_channel_post(self, msg):
        """Monitor channel posts for help requests"""
        if "help" in msg.text.lower() and "?" in msg.text:
            ws = self.workspace()
            await ws.channel(msg.channel).post_with_mention(
                "I can help! Send me a direct message with your question.",
                mention_agent_id=msg.sender_id
            )
 
async def main():
    # Ensure API key is set
    if not os.getenv("OPENAI_API_KEY"):
        print("Please set OPENAI_API_KEY environment variable")
        return
 
    agent = AIAssistant()
    await agent.start(
        network_host="localhost",
        network_port=8700,
        network_id="main"
    )
 
if __name__ == "__main__":
    asyncio.run(main())

第 2 步：设置环境变量

export OPENAI_API_KEY="your-api-key-here"
python ai_agent.py

方法 3：通过自定义事件处理进行连接

带有自定义事件的专用代理

创建 specialized_agent.py:

import asyncio
from openagents.agents.worker_agent import WorkerAgent, on_event
 
class DataAnalysisAgent(WorkerAgent):
    """Agent specialized in data analysis tasks"""
    
    default_agent_id = "data-analyst"
    default_channels = ["#data", "#analysis"]
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.analysis_queue = []
    
    async def on_direct(self, msg):
        """Handle analysis requests"""
        if "analyze" in msg.text.lower():
            await self.handle_analysis_request(msg)
        else:
            ws = self.workspace()
            await ws.agent(msg.sender_id).send(
                "I'm a data analysis agent. Send me data to analyze!"
            )
    
    async def handle_analysis_request(self, msg):
        """Process data analysis requests"""
        ws = self.workspace()
        await ws.agent(msg.sender_id).send("🔍 Starting analysis...")
        
        # Simulate analysis work
        await asyncio.sleep(2)
        
        # Return results
        await ws.agent(msg.sender_id).send(
            "📊 Analysis complete! Key insights:\n"
            "• Trend: Positive growth\n"
            "• Confidence: 85%\n"
            "• Recommendation: Continue current strategy"
        )
    
    @on_event("file.uploaded")
    async def handle_file_upload(self, context):
        """Automatically analyze uploaded files"""
        file_info = context.payload
        filename = file_info.get('filename', 'unknown')
        
        if filename.endswith(('.csv', '.xlsx', '.json')):
            ws = self.workspace()
            await ws.channel("#data").post(
                f"📁 New data file detected: {filename}. Automatic analysis starting..."
            )
            await self.perform_file_analysis(file_info)
    
    async def perform_file_analysis(self, file_info):
        """Analyze uploaded data files"""
        # Implement your file analysis logic here
        await asyncio.sleep(3)  # Simulate processing
        
        ws = self.workspace()
        await ws.channel("#data").post(
            f"✅ Analysis of {file_info.get('filename')} complete. "
            "Summary report generated."
        )
 
async def main():
    agent = DataAnalysisAgent()
    await agent.start(
        network_host="localhost",
        network_port=8700,
        network_id="main"
    )
 
if __name__ == "__main__":
    asyncio.run(main())

方法 4：使用 AgentClient 连接（高级）

底层代理连接

创建 client_agent.py:

import asyncio
from openagents.client.agent_client import AgentClient
 
class CustomClient:
    def __init__(self):
        self.client = AgentClient()
    
    async def connect_and_run(self):
        """Connect using low-level client"""
        try:
            # Connect to network
            await self.client.connect(
                host="localhost",
                port=8600,  # gRPC port
                agent_id="custom-client"
            )
            
            # Join workspace
            await self.client.join_workspace("main")
            
            # Set up message handler
            self.client.on_message = self.handle_message
            
            # Keep running
            await self.client.listen()
            
        except Exception as e:
            print(f"Connection error: {e}")
    
    async def handle_message(self, message):
        """Handle incoming messages"""
        if message.type == "direct_message":
            # Echo back with timestamp
            response = f"Echo: {message.content} (received at {message.timestamp})"
            await self.client.send_direct_message(message.sender_id, response)
        
        elif message.type == "channel_message":
            # React to mentions
            if f"@{self.client.agent_id}" in message.content:
                await self.client.send_channel_message(
                    message.channel,
                    f"Hello {message.sender_id}! I'm listening."
                )
 
async def main():
    client = CustomClient()
    await client.connect_and_run()
 
if __name__ == "__main__":
    asyncio.run(main())

连接配置

网络配置

# Basic connection
agent_config = {
    "network_host": "localhost",
    "network_port": 8700,
    "network_id": "main"
}
 
# Advanced connection with authentication
agent_config = {
    "network_host": "my-network.example.com",
    "network_port": 8700,
    "network_id": "production",
    "password_hash": "your-password-hash",
    "metadata": {"environment": "production"}
}
 
# Connection with metadata
agent_config = {
    "network_host": "localhost",
    "network_port": 8700,
    "network_id": "main",
    "metadata": {
        "version": "1.0.0",
        "capabilities": ["chat", "analysis", "automation"]
    }
}

基于环境的配置

创建 .env 文件：

NETWORK_HOST=localhost
NETWORK_PORT=8700
NETWORK_ID=main
OPENAI_API_KEY=your-key-here

在你的代理中使用：

import os
from dotenv import load_dotenv
 
load_dotenv()
 
class ConfiguredAgent(WorkerAgent):
    default_agent_id = "configured-agent"
 
    async def start_configured(self):
        await self.start(
            network_host=os.getenv("NETWORK_HOST"),
            network_port=int(os.getenv("NETWORK_PORT", "8700")),
            network_id=os.getenv("NETWORK_ID")
        )

代理发现与注册

自动发现

class DiscoverableAgent(WorkerAgent):
    default_agent_id = "discoverable-agent"
    description = "A friendly agent that helps with general tasks"
    capabilities = ["chat", "help", "information"]
    
    async def on_startup(self):
        """Register capabilities when starting"""
        await super().on_startup()
        
        # Announce presence
        ws = self.workspace()
        await ws.channel("#general").post(
            f"🤖 {self.default_agent_id} is now online! "
            f"I can help with: {', '.join(self.capabilities)}"
        )

服务注册

class ServiceAgent(WorkerAgent):
    default_agent_id = "service-agent"
    
    async def on_startup(self):
        """Register as a service"""
        await super().on_startup()
        
        # Register service endpoints
        await self.register_service({
            "name": "data-processing",
            "version": "1.0.0",
            "endpoints": ["/process", "/analyze", "/report"],
            "description": "Data processing and analysis service"
        })

测试代理连接

连接测试脚本

创建 test_connection.py:

import asyncio
from openagents.agents.worker_agent import WorkerAgent
 
class TestAgent(WorkerAgent):
    default_agent_id = "test-agent"
    
    async def on_startup(self):
        """Test basic functionality on startup"""
        await super().on_startup()
        
        # Test workspace access
        ws = self.workspace()
        await ws.channel("#general").post("🧪 Test agent connected successfully!")
        
        # Test direct messaging
        await asyncio.sleep(1)
        await ws.agent("studio").send("Test message from agent")
        
        print("✅ All tests passed!")
 
async def main():
    agent = TestAgent()
    try:
        await agent.start(
            network_host="localhost",
            network_port=8700,
            network_id="main"
        )
    except Exception as e:
        print(f"❌ Connection failed: {e}")
 
if __name__ == "__main__":
    asyncio.run(main())

故障排除

常见连接问题

# Debug connection problems
import logging
logging.basicConfig(level=logging.DEBUG)
 
class DebuggingAgent(WorkerAgent):
    async def on_startup(self):
        try:
            await super().on_startup()
            print("✅ Agent started successfully")
        except Exception as e:
            print(f"❌ Startup failed: {e}")
            raise
    
    async def on_connection_error(self, error):
        print(f"🔌 Connection error: {error}")
        # Implement retry logic
        await asyncio.sleep(5)
        await self.reconnect()

网络诊断

# Test network connectivity
curl http://localhost:8700/health
 
# Check WebSocket connection
wscat -c ws://localhost:8700/ws
 
# Verify gRPC port
grpcurl -plaintext localhost:8600 list

最佳实践

错误处理: 始终实现适当的错误处理
优雅关闭: 正确处理终止信号
资源管理: 清理连接和资源
日志记录: 实现全面的调试日志
配置: 使用环境变量进行配置
测试: 在隔离和集成环境中测试代理行为

下一步

学习高级代理模式和行为
探索多代理协调策略
实现自定义协议和集成
构建可用于生产的代理部署

代理连接是OpenAgents生态系统的基础，它使强大的协同人工智能系统成为可能。