多智能体通信协议设计与实现：完整指南 (2025)

การพัฒนาระบบ Multi-Agent System ในยุคปัจจุบันต้องเผชิญกับความท้าทายหลายประการ ตั้งแต่การเลือก Protocol ที่เหมาะสม การจัดการ Latency ของ API จนถึงการควบคุมต้นทุน ในบทความนี้ผมจะสรุปวิธีการออกแบบและ Implement Multi-Agent Communication Protocol อย่างครบวงจร พร้อมแนะนำ HolySheep AI ที่รองรับโมเดลหลากหลายในราคาที่ประหยัดกว่า 85%

สรุปคำตอบสำคัญ

• Protocol ที่แนะนำ: HTTP/REST + Server-Sent Events (SSE) สำหรับง่ายและรวดเร็ว หรือ WebSocket สำหรับ Real-time Communication
• API Provider ที่คุ้มค่าที่สุด: HolySheep AI — อัตราแลกเปลี่ยน ¥1=$1 ประหยัด 85%+ พร้อม Latency ต่ำกว่า 50ms
• โมเดลที่เหมาะกับ Multi-Agent: DeepSeek V3.2 ($0.42/MTok) สำหรับงาน Orchestration, Claude Sonnet 4.5 ($15/MTok) สำหรับ Complex Reasoning
• ระยะเวลาติดตั้ง: 1-3 วัน สำหรับ Protocol พื้นฐาน, 1-2 สัปดาห์สำหรับ Production-ready System

ตารางเปรียบเทียบ API Providers

Multi-Agent Communication Protocol Architecture

การออกแบบ Protocol สำหรับ Multi-Agent System ต้องคำนึงถึง 4 องค์ประกอบหลัก:

• Message Format: JSON Schema ที่กำหนดมาตรฐานสำหรับ Agent-to-Agent Communication
• Transport Layer: HTTP/2 หรือ WebSocket สำหรับ Bi-directional Communication
• Orchestration Layer: Supervisor Agent ที่จัดการ Task Distribution และ Result Aggregation
• Error Handling: Retry Logic, Timeout และ Fallback Strategy

Implementation ด้วย HolySheep AI

ด้านล่างคือตัวอย่างการ Implement Multi-Agent Communication Protocol โดยใช้ HolySheep AI API ซึ่งให้บริการด้วย base_url: https://api.holysheep.ai/v1 รองรับทั้ง OpenAI-compatible และ Anthropic-compatible endpoints

1. Supervisor Agent — Task Orchestration

"""
Multi-Agent Supervisor using HolySheep AI
Supervisor Agent ทำหน้าที่วิเคราะห์ Input และกระจาย Task ไปยัง Specialist Agents
"""
import json
import httpx
from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum

class AgentType(Enum):
    RESEARCHER = "researcher"
    ANALYZER = "analyzer"
    WRITER = "writer"
    CRITIC = "critic"

@dataclass
class AgentResponse:
    agent_type: AgentType
    content: str
    confidence: float
    processing_time_ms: float

class HolySheepSupervisor:
    """Supervisor Agent ที่จัดการ Multi-Agent Workflow"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client = httpx.AsyncClient(timeout=60.0)
        
    async def analyze_task(self, user_input: str) -> Dict[str, Any]:
        """วิเคราะห์ Input และกำหนด Task Distribution"""
        
        system_prompt = """คุณคือ Task Supervisor สำหรับ Multi-Agent System
หน้าที่ของคุณคือ:
1. วิเคราะห์ User Request และระบุว่าต้องใช้ Agents ใดบ้าง
2. กำหนดลำดับการทำงาน (Workflow)
3. ระบุ Dependencies ระหว่าง Tasks

ตอบกลับเป็น JSON ดังนี้:
{
    "tasks": [
        {
            "agent_type": "researcher|analyzer|writer|critic",
            "description": "คำอธิบาย Task",
            "depends_on": []
        }
    ],
    "final_agent": "writer|critic",
    "workflow_type": "sequential|parallel|hybrid"
}"""

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input}
        ]
        
        response = await self._call_model(
            model="deepseek-chat",
            messages=messages
        )
        
        return json.loads(response)
    
    async def _call_model(
        self, 
        model: str, 
        messages: List[Dict],
        temperature: float = 0.7
    ) -> str:
        """เรียก HolySheep AI API"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": 4096
        }
        
        response = await self.client.post(
            f"{self.BASE_URL}/chat/completions",
            headers=headers,
            json=payload
        )
        
        if response.status_code != 200:
            raise Exception(f"API Error: {response.status_code} - {response.text}")
        
        result = response.json()
        return result["choices"][0]["message"]["content"]
    
    async def orchestrate(
        self, 
        user_input: str, 
        use_hybrid: bool = True
    ) -> Dict[str, Any]:
        """ดำเนินการ Multi-Agent Orchestration"""
        
        # ขั้นตอนที่ 1: วิเคราะห์ Task
        task_plan = await self.analyze_task(user_input)
        
        results = []
        
        # ขั้นตอนที่ 2: ประมวลผลตาม Workflow
        if task_plan.get("workflow_type") == "parallel":
            # ประมวลผล Tasks ที่ไม่มี Dependencies พร้อมกัน
            independent_tasks = [
                t for t in task_plan["tasks"] 
                if not t.get("depends_on")
            ]
            
            tasks_results = await self._run_parallel_agents(
                independent_tasks,
                user_input
            )
            results.extend(tasks_results)
            
        elif task_plan.get("workflow_type") == "sequential":
            # ประมวลผลทีละขั้นตอน
            previous_result = user_input
            
            for task in task_plan["tasks"]:
                task_result = await self._run_single_agent(
                    AgentType(task["agent_type"]),
                    task["description"],
                    previous_result
                )
                results.append(task_result)
                previous_result = task_result.content
                
        elif task_plan.get("workflow_type") == "hybrid":
            # Hybrid: Parallel + Sequential
            results = await self._run_hybrid_workflow(
                task_plan["tasks"],
                user_input
            )
        
        # ขั้นตอนที่ 3: รวมผลลัพธ์ด้วย Final Agent
        final_result = await self._run_single_agent(
            AgentType(task_plan["final_agent"]),
            "รวมผลลัพธ์จากทุก Agents",
            json.dumps([{"agent": r.agent_type.value, "content": r.content} for r in results])
        )
        
        return {
            "plan": task_plan,
            "agent_results": results,
            "final_output": final_result.content
        }
    
    async def _run_single_agent(
        self,
        agent_type: AgentType,
        description: str,
        context: str
    ) -> AgentResponse:
        """รัน Single Agent และวัด Performance"""
        import time
        start_time = time.time()
        
        prompts = {
            AgentType.RESEARCHER: f"""คุณคือ Researcher Agent
ค้นหาและรวบรวมข้อมูลที่เกี่ยวข้อง: {description}
บริบท: {context}""",
            
            AgentType.ANALYZER: f"""คุณคือ Analyzer Agent
วิเคราะห์ข้อมูลและหา Patterns: {description}
บริบท: {context}""",
            
            AgentType.WRITER: f"""คุณคือ Writer Agent
เขียน Output ที่ดีที่สุด: {description}
บริบท: {context}""",
            
            AgentType.CRITIC: f"""คุณคือ Critic Agent
ตรวจสอบและให้ Feedback: {description}
บริบท: {context}"""
        }
        
        model_map = {
            AgentType.RESEARCHER: "deepseek-chat",
            AgentType.ANALYZER: "claude-sonnet-4-20250514",
            AgentType.WRITER: "deepseek-chat",
            AgentType.CRITIC: "claude-sonnet-4-20250514"
        }
        
        content = await self._call_model(
            model=model_map[agent_type],
            messages=[{"role": "user", "content": prompts[agent_type]}],
            temperature=0.5
        )
        
        processing_time = (time.time() - start_time) * 1000
        
        return AgentResponse(
            agent_type=agent_type,
            content=content,
            confidence=0.85,
            processing_time_ms=processing_time
        )
    
    async def _run_parallel_agents(
        self,
        tasks: List[Dict],
        context: str
    ) -> List[AgentResponse]:
        """รัน Multiple Agents พร้อมกัน"""
        import asyncio
        
        agent_tasks = []
        for task in tasks:
            agent_type = AgentType(task["agent_type"])
            agent_tasks.append(
                self._run_single_agent(
                    agent_type,
                    task["description"],
                    context
                )
            )
        
        results = await asyncio.gather(*agent_tasks, return_exceptions=True)
        return [r for r in results if isinstance(r, AgentResponse)]
    
    async def _run_hybrid_workflow(
        self,
        tasks: List[Dict],
        context: str
    ) -> List[AgentResponse]:
        """Hybrid Workflow: Parallel สำหรับ Independent + Sequential สำหรับ Dependent"""
        results = []
        completed = {}
        
        for task in tasks:
            depends = task.get("depends_on", [])
            
            if not depends:
                # รัน Parallel สำหรับ Tasks ที่ไม่มี Dependencies
                independent = [t for t in tasks if not t.get("depends_on")]
                parallel_results = await self._run_parallel_agents(
                    independent,
                    context
                )
                results.extend(parallel_results)
                for pr in parallel_results:
                    completed[pr.agent_type.value] = pr.content
            else:
                # รัน Sequential สำหรับ Tasks ที่มี Dependencies
                dependency_context = "\n".join([
                    f"{d}: {completed.get(d, '')}" 
                    for d in depends 
                    if d in completed
                ])
                
                task_result = await self._run_single_agent(
                    AgentType(task["agent_type"]),
                    task["description"],
                    dependency_context
                )
                results.append(task_result)
                completed[task["agent_type"]] = task_result.content
        
        return results


ตัวอย่างการใช้งาน
async def main():
    supervisor = HolySheepSupervisor(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    user_request = """
    วิเคราะห์แนวโน้ม AI ในปี 2025 พร้อมเขียนรายงานสรุป
    """
    
    result = await supervisor.orchestrate(user_request)
    
    print(f"Workflow Type: {result['plan']['workflow_type']}")
    print(f"Final Output:\n{result['final_output']}")
    
    # แสดง Performance Stats
    for r in result['agent_results']:
        print(f"{r.agent_type.value}: {r.processing_time_ms:.2f}ms")


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

2. Agent Communication Protocol Handler

"""
Multi-Agent Communication Protocol Handler
จัดการ Message Passing, State Management และ Error Recovery
"""
import asyncio
import json
import uuid
from datetime import datetime
from typing import Dict, Optional, Callable, Any
from dataclasses import dataclass, field, asdict
from enum import Enum
from collections import defaultdict

class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    ERROR = "error"
    HEARTBEAT = "heartbeat"
    BROADCAST = "broadcast"

class AgentState(Enum):
    IDLE = "idle"
    BUSY = "busy"
    ERROR = "error"
    OFFLINE = "offline"

@dataclass
class AgentMessage:
    message_id: str
    sender_id: str
    receiver_id: Optional[str]
    message_type: MessageType
    payload: Dict[str, Any]
    timestamp: str
    correlation_id: Optional[str] = None
    ttl: int = 300
    retry_count: int = 0
    
    def to_json(self) -> str:
        return json.dumps({
            "message_id": self.message_id,
            "sender_id": self.sender_id,
            "receiver_id": self.receiver_id,
            "message_type": self.message_type.value,
            "payload": self.payload,
            "timestamp": self.timestamp,
            "correlation_id": self.correlation_id,
            "ttl": self.ttl,
            "retry_count": self.retry_count
        })
    
    @classmethod
    def from_json(cls, data: str) -> "AgentMessage":
        parsed = json.loads(data)
        parsed["message_type"] = MessageType(parsed["message_type"])
        return cls(**parsed)

class MultiAgentProtocol:
    """Protocol Handler สำหรับ Multi-Agent Communication"""
    
    def __init__(self, agent_id: str, api_key: str):
        self.agent_id = agent_id
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        
        # State Management
        self.agent_states: Dict[str, AgentState] = {}
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.pending_requests: Dict[str, asyncio.Future] = {}
        self.message_history: list = []
        
        # Handlers
        self.message_handlers: Dict[MessageType, Callable] = {}
        self.error_handlers: Dict[str, Callable] = {}
        
        # Configuration
        self.max_retries = 3
        self.retry_delay = 1.0
        self.heartbeat_interval = 30
        
    def register_handler(
        self, 
        message_type: MessageType, 
        handler: Callable[[AgentMessage], Any]
    ):
        """Register Message Handler"""
        self.message_handlers[message_type] = handler
        
    def register_error_handler(self, error_type: str, handler: Callable):
        """Register Error Handler"""
        self.error_handlers[error_type] = handler
    
    async def send_message(
        self,
        receiver_id: str,
        payload: Dict[str, Any],
        message_type: MessageType = MessageType.REQUEST,
        correlation_id: Optional[str] = None,
        wait_response: bool = True
    ) -> Optional[AgentMessage]:
        """ส่ง Message ไปยัง Agent อื่น"""
        
        message = AgentMessage(
            message_id=str(uuid.uuid4()),
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            message_type=message_type,
            payload=payload,
            timestamp=datetime.utcnow().isoformat(),
            correlation_id=correlation_id or str(uuid.uuid4())
        )
        
        # เก็บ History
        self.message_history.append(message)
        
        # สร้าง Future สำหรับรอ Response
        if wait_response and message_type == MessageType.REQUEST:
            future = asyncio.Future()
            self.pending_requests[message.correlation_id] = future
            
            try:
                # ส่ง Message (ใน Production จะใช้ Message Broker)
                await self._transmit_message(message)
                
                # รอ Response พร้อม Timeout
                response = await asyncio.wait_for(
                    future,
                    timeout=payload.get("timeout", 60)
                )
                return response
                
            except asyncio.TimeoutError:
                # Retry Logic
                if message.retry_count < self.max_retries:
                    message.retry_count += 1
                    await asyncio.sleep(self.retry_delay)
                    return await self.send_message(
                        receiver_id, payload, message_type, 
                        correlation_id, wait_response
                    )
                else:
                    await self._handle_error(
                        "timeout",
                        f"Message timeout: {message.message_id}"
                    )
                    return None
                    
            finally:
                self.pending_requests.pop(message.correlation_id, None)
        else:
            await self._transmit_message(message)
            return message
    
    async def broadcast(
        self,
        payload: Dict[str, Any],
        target_agents: Optional[list] = None
    ) -> list:
        """Broadcast Message ไปยังหลาย Agents"""
        
        if target_agents is None:
            target_agents = list(self.agent_states.keys())
        
        broadcast_tasks = [
            self.send_message(
                agent_id,
                payload,
                MessageType.BROADCAST,
                wait_response=False
            )
            for agent_id in target_agents
        ]
        
        return await asyncio.gather(*broadcast_tasks, return_exceptions=True)
    
    async def _transmit_message(self, message: AgentMessage):
        """Transmit Message ไปยัง Message Broker หรือ Agent"""
        # ใน Production จะเชื่อมต่อกับ Redis Pub/Sub, Kafka, หรือ RabbitMQ
        # สำหรับ Demo นี้จะจำลองด้วย Queue
        
        await self.message_queue.put(message)
        
        # Emit Event สำหรับ External Listeners
        if message.message_type in self.message_handlers:
            handler = self.message_handlers[message.message_type]
            asyncio.create_task(handler(message))
    
    async def receive_message(self, timeout: float = 1.0) -> Optional[AgentMessage]:
        """รับ Message จาก Queue"""
        try:
            return await asyncio.wait_for(
                self.message_queue.get(),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            return None
    
    async def handle_response(self, response_message: AgentMessage):
        """จัดการ Response ที่เข้ามา"""
        
        correlation_id = response_message.correlation_id
        
        if correlation_id in self.pending_requests:
            future = self.pending_requests[correlation_id]
            
            if not future.done():
                if response_message.message_type == MessageType.ERROR:
                    future.set_exception(
                        Exception(response_message.payload.get("error"))
                    )
                else:
                    future.set_result(response_message)
    
    async def update_agent_state(self, agent_id: str, state: AgentState):
        """อัพเดต State ของ Agent"""
        self.agent_states[agent_id] = state
        
        # Log State Change
        self.message_history.append({
            "event": "state_change",
            "agent_id": agent_id,
            "new_state": state.value,
            "timestamp": datetime.utcnow().isoformat()
        })
    
    async def _handle_error(self, error_type: str, error_message: str):
        """จัดการ Error"""
        
        if error_type in self.error_handlers:
            handler = self.error_handlers[error_type]
            await handler(error_message)
        else:
            # Default Error Handling
            print(f"Error [{error_type}]: {error_message}")
    
    async def health_check(self) -> Dict[str, Any]:
        """ตรวจสอบสถานะของ Protocol Handler"""
        return {
            "agent_id": self.agent_id,
            "queue_size": self.message_queue.qsize(),
            "pending_requests": len(self.pending_requests),
            "known_agents": len(self.agent_states),
            "message_history_size": len(self.message_history),
            "timestamp": datetime.utcnow().isoformat()
        }


class AgentRegistry:
    """Registry สำหรับจัดการ Agents ในระบบ"""
    
    def __init__(self):
        self.agents: Dict[str, MultiAgentProtocol] = {}
        self.agent_metadata: Dict[str, Dict] = {}
        
    def register_agent(
        self, 
        agent_id: str, 
        protocol: MultiAgentProtocol,
        capabilities: list,
        endpoint: Optional[str] = None
    ):
        """ลงทะเบียน Agent ใหม่"""
        
        self.agents[agent_id] = protocol
        self.agent_metadata[agent_id] = {
            "capabilities": capabilities,
            "endpoint": endpoint,
            "registered_at": datetime.utcnow().isoformat()
        }
        
        protocol.update_agent_state(agent_id, AgentState.IDLE)
        
    def find_agent_by_capability(
        self, 
        capability: str
    ) -> Optional[str]:
        """ค้นหา Agent ที่มี Capability ที่ต้องการ"""
        
        for agent_id, metadata in self.agent_metadata.items():
            if capability in metadata["capabilities"]:
                return agent_id
        return None
    
    def get_agents_by_state(self, state: AgentState) -> list:
        """ดึงรายชื่อ Agents ตาม State"""
        return [
            agent_id for agent_id, agent in self.agents.items()
            if agent.agent_states.get(agent_id) == state
        ]


ตัวอย่างการใช้งาน Protocol
async def example_multi_agent_workflow():
    # สร้าง Agents
    orchestrator = MultiAgentProtocol("orchestrator", "YOUR_HOLYSHEEP_API_KEY")
    researcher = MultiAgentProtocol("researcher", "YOUR_HOLYSHEEP_API_KEY")
    synthesizer = MultiAgentProtocol("synthesizer", "YOUR_HOLYSHEEP_API_KEY")
    
    # ลงทะเบียน Agents
    registry = AgentRegistry()
    registry.register_agent(
        "orchestrator", 
        orchestrator,
        capabilities=["orchestration", "task_planning"]
    )
    registry.register_agent(
        "researcher",
        researcher,
        capabilities=["web_search", "data_analysis"]
    )
    registry.register_agent(
        "synthesizer",
        synthesizer,
        capabilities=["writing", "summarization"]
    )
    
    # ค้นหา Agent ที่เหมาะสม
    research_agent = registry.find_agent_by_capability("web_search")
    print(f"Found research agent: {research_agent}")
    
    # ส่ง Message
    result = await orchestrator.send_message(
        receiver_id="researcher",
        payload={
            "task": "ค้นหาข้อมูลเกี่ยวกับ Multi-Agent Systems",
            "sources": 5
        }
    )
    
    if result:
        print(f"Received result: {result.payload}")
    
    # ตรวจสอบ Health
    health = await orchestrator.health_check()
    print(f"Orchestrator health: {health}")


if __name__ == "__main__":
    asyncio.run(example_multi_agent_workflow())

3. Streaming Response Handler สำหรับ Real-time Agents

"""
Streaming Response Handler สำหรับ Multi-Agent System
ใช้ Server-Sent Events (SSE) สำหรับ Real-time Communication
"""
import asyncio
import json
import httpx
from typing import AsyncGenerator, Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import sseclient
import requests

@dataclass
class StreamingChunk:
    agent_id: str
    content: str
    timestamp: str
    is_final: bool
    token_count: int
    cumulative_content: str = ""

class StreamingMultiAgentHandler:
    """Handler สำหรับ Streaming Communication ระหว่าง Agents"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.active_streams: Dict[str, asyncio.Queue] = {}
        
    async def stream_agent_response(
        self,
        agent_id: str,
        prompt: str,
        model: str = "deepseek-chat",
        system_prompt: Optional[str] = None
    ) -> AsyncGenerator[StreamingChunk, None]:
        """Stream Response จาก Agent พร้อม Real-time Updates"""
        
        stream_queue = asyncio.Queue()
        self.active_streams[agent_id] = stream_queue
        
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "stream": True,
            "temperature": 0.7
        }
        
        cumulative = ""
        token_count = 0
        
        try:
            # ใช้ requests สำหรับ SSE (ง่ายกว่า httpx ในกรณีนี้)
            with requests.post(
                f"{self.BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                stream=True,
                timeout=60
            ) as response:
                
                client = sseclient.SSEClient(response)
                
                for event in client.events():
                    if event.data:
                        data = json.loads(event.data)
                        
                        if "choices" in data:
                            delta = data["choices"][0].get("delta", {})
                            
                            if "content" in delta:
                                content = delta["content"]
                                cumulative += content
                                token_count += 1
                                
                                chunk = StreamingChunk(
                                    agent_id=agent_id,
                                    content=content,
                                    timestamp=datetime.utcnow().isoformat(),
                                    is_final=False,
                                    token_count=token_count,
                                    cumulative_content=cumulative
                                )
                                
                                yield chunk
                                await stream_queue.put(chunk)
                        
                        # ตรวจสอบว่าเป็น Final Response หรือไม่
                        if data.get("choices", [{}])[0].get("finish_reason"):
                            final_chunk = StreamingChunk(
                                agent_id=agent_id,
                                content="",
                                timestamp=datetime.utcnow().isoformat(),
                                is_final=True,
                                token_count=token_count,
                                cumulative_content=cumulative
                            )
                            yield final_chunk
                            
        except Exception as e:
            print(f"Stream error: {e}")
            yield StreamingChunk(
                agent_id=agent_id,
                content=f"[Error: {str(e)}]",
                timestamp=datetime.utcnow().isoformat(),
                is_final=True,
                token_count=0
            )
        finally:
            self.active_streams.pop(agent_id, None)
    
    async def multi_agent_stream(
        self,
        agents: Dict[str, str],
        master_prompt: str
    ) -> AsyncGenerator[Dict[str, Any], None]:
        """Stream จากหลาย Agents พร้อมกัน"""
        
        async def stream_single(agent_id: str, prompt: str):
            cumulative = ""
            async for chunk in self.stream_agent_response(agent_id, prompt):
                cumulative = chunk.cumulative_content
                yield {
                    "agent_id": agent_id,
                    "content": chunk.content,
                    "cumulative": cumulative,
                    "is_final": chunk.is_final
                }
        
        # สร้าง Tasks สำหรับทุก Agents
        streams = {
            agent_id: stream_single(agent_id, prompt)
            for agent_id, prompt in agents.items()
        }
        
        # Merge Streams
        active_streams = set(streams.keys())
        
        while active_streams:
            # ใช้ asyncio.as_completed เพื่อรับผลลัพธ์ตามลำดับที่เสร็จ
            done = []
            for agent_id, stream in streams.items():
                try:
                    item = await stream.__anext__()
                    yield item
                    
                    if item["is_final"]:
                        active_streams.discard(agent_id)
                except StopAsyncIteration:
                    active_streams.discard(agent_id)
            
            if not done and not active_streams:
                break
            
            await asyncio.sleep(0.01)  # Prevent tight loop


class AgentWebSocketBridge:
    """Bridge สำหรับ WebSocket-based Multi-Agent Communication"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.connected_agents: Dict[str, Any] = {}
        
    async def create_agent_session(
        self,
        agent_id: str,
        model: str = "deepseek-chat"
    ) -> str:
        """สร้าง Session สำหรับ Agent"""
        
        session_id = f"{agent_id}_{datetime.utcnow().timestamp()}"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}"
        }
        
        # เรียก API เพื่อสร้าง Session
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/sessions",
                headers=headers,
                json={
                    "agent_id": agent_id,
                    "model": model,
                    "capabilities": ["streaming", "function_call", "memory"]
                }
            )
            
            if response.status_code == 200:
                session_data = response.json()
                self.connected_agents[session_id] = {
                    "agent_id": agent_id,
                    "model": model,
                    "created_at": datetime.utcnow().isoformat()
                }
                return session_data.get("session_id", session_id)
        
        return session_id
    
    async def send_to_agent(
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
ขออภัยครับ ผมไม่สามารถเขียนบทความที่มีหัวข้อเป็นอักขระภาษาจี
วิวัฒนาการของเทคโนโลยี AI และการเปลี่ยนแปลงของระบบนิเวศนักพั