A2A उदाहरण: Travel Planner OpenRouter

यह एक Python कार्यान्वयन है जो A2A (Agent2Agent) प्रोटोकॉल का पालन करता है। Google के आधिकारिक a2a-python SDK और OpenAI Python SDK के आधार पर कार्यान्वित एक यात्रा सहायक डेमो। यह OpenAI मॉडल विनिर्देशों के अनुरूप एक यात्रा सहायक है, जो आपको यात्रा योजना सेवाएं प्रदान करने में सक्षम है।

स्रोत कोड

A2A Travel Planner OpenRouter

परियोजना आर्किटेक्चर

यह परियोजना A2A प्रोटोकॉल का उपयोग करके एक अंतर-संचालनीय यात्रा योजना एजेंट बनाने का तरीका दिखाती है, जिसमें निम्नलिखित मुख्य घटक शामिल हैं:

Travel Planner Agent: OpenAI-संगत इंटरफेस पर आधारित मुख्य यात्रा सहायक तर्क
Agent Executor: A2A प्रोटोकॉल एडाप्टर जो एजेंट तर्क को A2A सर्वर से जोड़ता है
A2A Server: A2A प्रोटोकॉल-अनुपालित सर्वर जो मानकीकृत अंतर-एजेंट संचार इंटरफेस प्रदान करता है
Loop Client: A2A सर्वर के साथ बातचीत के लिए परीक्षण क्लाइंट

वर्कफ़्लो अनुक्रम आरेख

sequenceDiagram
    participant Client
    participant A2AServer
    participant RequestHandler
    participant Executor as TravelPlannerAgentExecutor
    participant Agent as TravelPlannerAgent
    participant LLM as OpenAI-Compatible LLM

    Client->>A2AServer: एजेंट कार्ड का अनुरोध
    A2AServer->>Client: एजेंट कार्ड वापस करें (कौशल, क्षमताएं)

    Note over Client,A2AServer: उपयोगकर्ता यात्रा योजना की क्वेरी करता है

    Client->>A2AServer: message/sendStream (स्ट्रीमिंग अनुरोध)
    A2AServer->>RequestHandler: स्ट्रीमिंग अनुरोध को रूट करें
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop स्ट्रीमिंग प्रतिक्रिया प्रसंस्करण
        LLM-->>Agent: स्ट्रीमिंग सामग्री खंड वापस करें
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: SSE इवेंट पुश करें
        A2AServer-->>Client: स्ट्रीम सामग्री अपडेट
    end
    
    LLM-->>Agent: अंतिम प्रतिक्रिया पूर्ण
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: अंतिम TaskArtifactUpdateEvent
    RequestHandler-->>A2AServer: अंतिम SSE इवेंट
    A2AServer-->>Client: स्ट्रीमिंग प्रतिक्रिया पूर्ण

मुख्य वर्कफ़्लो

एजेंट कार्ड पुनर्प्राप्ति: क्लाइंट पहले A2A सर्वर से एजेंट कार्ड प्राप्त करता है ताकि एजेंट की क्षमताओं और कौशल को समझ सके
स्ट्रीमिंग अनुरोध प्रसंस्करण: क्लाइंट उपयोगकर्ता क्वेरी के साथ स्ट्रीमिंग संदेश अनुरोध भेजता है
एजेंट निष्पादन: एजेंट निष्पादक अनुरोध को संसाधित करता है और यात्रा योजनाकार एजेंट मुख्य तर्क को कॉल करता है
LLM इंटरैक्शन: एजेंट OpenAI-संगत LLM के साथ स्ट्रीमिंग बातचीत करता है
वास्तविक समय प्रतिक्रिया: Server-Sent Events (SSE) के माध्यम से क्लाइंट को वास्तविक समय में प्रतिक्रियाएं स्ट्रीम करता है

शुरुआत करना

पर्यावरण चर कॉन्फ़िगर करें:

उदाहरण फ़ाइल कॉपी करें और अपनी API क्रेडेंशियल कॉन्फ़िगर करें।

cp env.example .env

अपने वास्तविक मानों के साथ .env फ़ाइल संपादित करें:

# आवश्यक: AI मॉडल सेवा के लिए आपकी API कुंजी
API_KEY=your_actual_api_key_here

# वैकल्पिक: मॉडल नाम (डिफ़ॉल्ट: google/gemini-2.0-flash-001)
MODEL_NAME=google/gemini-2.0-flash-001

# वैकल्पिक: API सेवा के लिए आधार URL
BASE_URL=https://openrouter.ai/api/v1

निर्भरताएं स्थापित करें और सर्वर शुरू करें:
```
uv venv
source .venv/bin/activate

uv sync
uv run .
```
नए टर्मिनल में लूप क्लाइंट चलाएं:
```
source .venv/bin/activate
uv run loop_client.py
```

कॉन्फ़िगरेशन

एप्लिकेशन कॉन्फ़िगरेशन के लिए पर्यावरण चर का उपयोग करता है:

API_KEY (आवश्यक): AI मॉडल सेवा के लिए आपकी API कुंजी
MODEL_NAME (वैकल्पिक): उपयोग करने के लिए मॉडल नाम (डिफ़ॉल्ट: "google/gemini-2.0-flash-001")
BASE_URL (वैकल्पिक): API सेवा के लिए आधार URL (डिफ़ॉल्ट: "https://openrouter.ai/api/v1")

तकनीकी विशेषताएं

वर्तमान कार्यान्वयन

✅ A2A प्रोटोकॉल अनुपालन: Agent2Agent प्रोटोकॉल विनिर्देश के साथ पूर्ण अनुपालन
✅ स्ट्रीमिंग प्रतिक्रिया: वास्तविक समय स्ट्रीमिंग सामग्री उत्पादन का समर्थन
✅ OpenAI संगत: किसी भी OpenAI-संगत API इंटरफेस का समर्थन
✅ मॉड्यूलर डिज़ाइन: एजेंट तर्क और प्रोटोकॉल अनुकूलन का स्पष्ट पृथक्करण
✅ पर्यावरण कॉन्फ़िगरेशन: लचीला पर्यावरण चर कॉन्फ़िगरेशन

भविष्य की वृद्धि योजनाएं

कार्य स्थिति प्रबंधन वृद्धि

Google A2A LangGraph नमूना के आधार पर, निम्नलिखित सुविधाएं जोड़ने की योजना है:

🔄 कार्य जीवनचक्र प्रबंधन: पूर्ण कार्य स्थिति ट्रैकिंग का कार्यान्वयन (सबमिट किया गया → काम कर रहा → पूर्ण/असफल)
🔄 बहु-मोड़ बातचीत समर्थन: जटिल यात्रा योजना परिदृश्यों का समर्थन करने के लिए input_required स्थिति जोड़ें जिसमें उपयोगकर्ता स्पष्टीकरण की आवश्यकता होती है
🔄 कार्य दृढ़ता: लंबे समय तक चलने वाले योजना कार्यों के लिए कार्य स्थिति दृढ़ता का कार्यान्वयन
🔄 बेहतर त्रुटि हैंडलिंग: अधिक विस्तृत त्रुटि स्थितियां और पुनर्प्राप्ति तंत्र
🔄 कार्य रद्दीकरण: चल रहे कार्यों के रद्दीकरण का समर्थन

स्थिति प्रबंधन उदाहरण

# भविष्य की स्थिति प्रबंधन कार्यान्वयन उदाहरण
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # पता लगाएं कि क्या अधिक जानकारी की आवश्यकता है
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="अधिक जानकारी की आवश्यकता: कृपया विशिष्ट गंतव्य, तारीखें और बजट सीमा प्रदान करें"
            )
        
        # जटिल बहु-चरणीय योजना निष्पादित करें
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="विस्तृत यात्रा योजना बना रहे हैं..."
        )

नियोजित सुविधा जोड़

📋 संरचित डेटा समर्थन: फॉर्म-आधारित यात्रा प्राथमिकता संग्रह के लिए DataPart समर्थन जोड़ें
🖼️ मल्टीमीडिया समर्थन: यात्रा छवियों, मानचित्रों आदि के उत्पादन और प्रसंस्करण के लिए FilePart समर्थन
🔍 उपकरण एकीकरण: बाहरी API (मौसम, उड़ानें, होटल आदि) को उपकरण कॉल के रूप में एकीकृत करें
🌐 बहुभाषी समर्थन: बहुभाषी यात्रा योजना क्षमताओं का विस्तार
📊 विश्लेषण मेट्रिक्स: कार्य निष्पादन समय, सफलता दर मेट्रिक्स का संग्रह जोड़ें

A2A पर जाएं