a2a/demo_no_a2a.py

"""
❌ 非 A2A 模式：两个 Agent 各自有完全不同的 API
需要手动适配，调用方必须提前知道每个 Agent 的接口细节
"""
from openai import AsyncOpenAI
import asyncio

DEEPSEEK_API_KEY = "你的APIKEY"
client = AsyncOpenAI(api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com/v1")


async def call_llm(system: str, user: str) -> str:
    r = await client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
        temperature=0.3, max_tokens=1024,
    )
    return r.choices[0].message.content


# ─── "Agent" A：问答服务 ─────────────────────────────
# 假装这是一个外部的问答 API，格式如下：
#   POST /ask  {"question": "..."} → {"answer": "..."}
# 调用方必须：知道 URL、知道字段名是 "question"、"answer"

async def agent_qa(question: str) -> str:
    """模拟调用一个自定义问答 API"""
    print("  [适配] POST /ask  body={\"question\": \"...\"}")
    return await call_llm(
        "你是一个有用的AI助手",
        question,
    )


# ─── "Agent" B：翻译服务 ─────────────────────────────
# 假装这是另一个公司的翻译 API，格式完全不一样：
#   POST /v2/translate  {"src": "...", "from": "zh", "to": "en"}
#   → {"code": 0, "data": {"result": "..."}}
# 调用方必须：知道不同 URL、不同字段名、不同响应结构

async def agent_translate(text: str) -> str:
    """模拟调用一个自定义翻译 API"""
    print("  [适配] POST /v2/translate  body={\"src\": \"...\", \"from\": \"zh\", \"to\": \"en\"}")
    print("  [适配] 解析响应: resp.data.result")
    return await call_llm(
        "你是一个专业中译英翻译，只输出翻译结果",
        f"把以下中文翻译成英文：{text}",
    )


# ─── 调用方：没有 A2A，必须手工适配每个 Agent ────

async def main():
    question = "什么是微服务架构？"

    print("=" * 64)
    print("  ❌ 没有 A2A 协议")
    print("  两个 Agent，两种 API，两套适配代码")
    print("=" * 64)

    print(f"\n  用户: {question}")

    # Step 1: 必须先知道问答 API 的地址和格式
    print("\n  ── 调用 问答 Agent ──")
    qa_result = await agent_qa(question)
    print(f"  回答: {qa_result[:80]}...")

    # Step 2: 想把回答翻译成英文
    #   → 又得知道翻译 API 的地址和格式（完全不同！）
    print("\n  ── 调用 翻译 Agent ──")
    print("  ⚠ 需要写全新的适配代码：不同 URL、不同字段名、不同响应结构")
    trans_result = await agent_translate(qa_result)
    print(f"  翻译: {trans_result[:80]}...")

    # Step 3: 如果想再加第三个 Agent（比如总结）
    #   → 又要第三套适配代码
    print("\n  ❌ 再加第3个 Agent → 再写第3套适配")
    print("  N 个 Agent = N 套适配代码（API 格式差异越大越痛苦）")
    print()

    # ===== 显示对比总结 =====
    print("-" * 64)
    print("  对比：")
    print("  问答 API: POST /ask  {\"question\": ...} -> {\"answer\": ...}")
    print("  翻译 API: POST /v2/translate  {\"src\": ...} -> {\"code\":0, \"data\":{\"result\":...}}")
    print("  → 完全不同的 URL、字段名、响应结构")
    print("  → 每接入一个 Agent 都要看文档写适配")


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