用 AIMock 讓 AI 代理測試變得決定性 — 在單一連接埠模擬 LLM、向量資料庫、MCP 與 A2A

import { LLMock, VectorMock } from "@copilotkit/aimock";

const llmock = new LLMock({ port: 0 }); // 隨機連接埠

// 依系統提示的內容分派測試固件
llmock.on({ systemMessage: "evaluating product offers" }, {
  content: JSON.stringify({
    topPickIndex: 0,
    reasoning: "The MacBook Air M4 offers the best combination...",
    confidence: 0.92,
  }),
});

llmock.on({ systemMessage: "purchasing concierge" }, {
  content: JSON.stringify({
    refinedQuery: "ultrabook laptop under 1.3kg...",
    priorities: ["weight under 1.3kg", "battery life 15+ hours"],
    priceRange: { min: 900, max: 1300 },
  }),
});

await llmock.start();
process.env.OPENAI_BASE_URL = `${llmock.url}/v1`;
process.env.OPENAI_API_KEY = "mock-key";

const vectorMock = new VectorMock({ port: 0 });

vectorMock.addCollection("product-knowledge", { dimension: 1536 });

// 註冊靜態結果：每次都回傳同樣的結果
vectorMock.onQuery("product-knowledge", [
  {
    id: "review-macbook-air",
    score: 0.94,
    metadata: {
      brand: "Apple",
      model: "MacBook Air M4",
      text: "MacBook Air M4: 1.24kg, 18hr battery life...",
    },
  },
  // ... 其他項目
]);

await vectorMock.start();

const server = createServer((req, res) => {
  // AI 代理卡片探索：GET /.well-known/agent-card.json (v1.0)
  if (req.url === "/.well-known/agent-card.json" && req.method === "GET") {
    res.writeHead(200, { "Content-Type": "application/json" });
    res.end(JSON.stringify({
      name: "Mock Electronics Seller",
      skills: [{ id: "product-search", name: "Product Search", ... }],
      ...
    }));
    return;
  }

  // JSON-RPC：POST /，SendMessage (v1.0)
  if (req.method === "POST") {
    const rpc = JSON.parse(body);
    if (rpc.method === "SendMessage") {
      res.end(JSON.stringify({
        jsonrpc: "2.0", id: rpc.id,
        result: {
          message: {
            role: "ROLE_AGENT",
            parts: [{ data: { offers: mockOffers } }],
            messageId: `reply-${rpc.id}`,
            contextId: rpc.params?.message?.contextId,
          },
        },
      }));
    }
  }
});

test("same query twice returns identical results (determinism proof)", async () => {
  const results1 = await kb.query("laptop");
  const results2 = await kb.query("laptop");

  // 真實的 Pinecone 沒有這種保證
  expect(results1).toEqual(results2);
});

test("works without RAG knowledge (graceful degradation)", async () => {
  const agent = new BuyerAgent({
    sellerAgentUrl: sellerUrl,
    knowledgeBase: {
      apiKey: "mock-key",
      baseUrl: "http://localhost:1", // 連不到 → 回退
      indexName: "product-knowledge",
    },
    ragTimeoutMs: 500,
  });

  const result = await agent.findProduct({
    query: "I need a lightweight laptop for travel",
  });

  // 即使 RAG 停擺，AI 代理仍能靠 LLM + A2A 運作
  expect(result.topPick).toBeDefined();
  expect(result.knowledgeUsed).toHaveLength(0);
});

requires-python = ">=3.13"

FROM python:3.13-slim

[[tool.uv.index]]
url = "https://pypi.org/simple/"
default = true

from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.types import AgentCard, AgentInterface, AgentCapabilities, AgentSkill
from starlette.applications import Starlette

agent_card = AgentCard(
    name="burger_seller_agent",
    description="Helps with creating burger orders",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True),
    supported_interfaces=[
        AgentInterface(protocol_binding="JSONRPC", url=agent_host_url)
    ],
    skills=[skill],
)

request_handler = DefaultRequestHandler(
    agent_executor=BurgerSellerAgentExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,  # v1.0 起為必填
)

routes = []
routes.extend(create_agent_card_routes(agent_card))
routes.extend(create_jsonrpc_routes(request_handler, "/"))

app = Starlette(routes=routes)
uvicorn.run(app, host=host, port=port)

from a2a.types import Message, Part, Role, Task
from a2a.utils.errors import UnsupportedOperationError

class BurgerSellerAgentExecutor(AgentExecutor):
    async def execute(self, context, event_queue):
        query = context.message.parts[0].text if context.message else ""
        result = await self.agent.invoke(query, context.context_id)

        # v1.0：直接用 Part(text="...") 建構，不需要 TextPart 包裝
        await event_queue.enqueue_event(
            Message(
                role=Role.ROLE_AGENT,
                parts=[Part(text=str(result))],
                message_id=context.task_id or "msg",
                context_id=context.context_id,
            )
        )

def invoke(self, query, sessionId) -> str:
async def invoke(self, query, sessionId) -> str:
    # ...
    response = crew.kickoff(inputs)
    response = await crew.kickoff_async(inputs)
    return response

export class A2AClient {
  // v1.0：agent-card.json（原為 agent.json）
  async discover(): Promise<AgentCard> {
    const url = `${this.agentUrl}/.well-known/agent-card.json`;
    const res = await fetch(url);
    this.card = await res.json();
    return this.card;
  }

  // v1.0：SendMessage 加上 A2A-Version 標頭
  async sendMessage(text: string, contextId?: string): Promise<A2AMessage> {
    const rpcRequest = {
      jsonrpc: "2.0",
      id: messageId,
      method: "SendMessage",  // 原為 "tasks/send"
      params: {
        message: {
          role: "ROLE_USER",  // 原為 "user"
          parts: [{ text }],  // 原為 { type: "text", text }
          messageId,
          contextId,
        },
      },
    };

    const res = await fetch(this.agentUrl, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "A2A-Version": "1.0",  // v1.0 起為必填
      },
      body: JSON.stringify(rpcRequest),
    });

    const rpcResponse = await res.json();
    return rpcResponse.result.message;  // 原為 result（任務物件）
  }
}

test("setChaos({ malformedRate: 1 }) — every response is garbled", async () => {
  const mock = new LLMock({ port: 0 });

  // 註冊有效的測試固件（/.*/ 會比對所有訊息）
  mock.onMessage(/.*/, {
    content: '{"refinedQuery": "test", "priorities": [], "priceRange": {"min": 0, "max": 100}}',
  });

  // 混沌：100% 破壞回應
  mock.setChaos({ malformedRate: 1.0 });

  await mock.start();
  process.env.OPENAI_BASE_URL = `http://localhost:${mock.port}/v1`;

  const reasoner = new Reasoner("gpt-4o-mini");

  // 測試固件有比對到，但混沌把回應弄壞了
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  await mock.stop();
});

test("clearChaos() restores normal operation", async () => {
  const mock = new LLMock({ port: 0 });
  mock.onMessage(/.*/, { content: validJson });
  mock.setChaos({ malformedRate: 1.0 });

  await mock.start();
  // ... 環境設定 ...

  const reasoner = new Reasoner("gpt-4o-mini");

  // 開啟混沌：失敗
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  // 關閉混沌：正常運作
  mock.clearChaos();

  const strategy = await reasoner.buildSearchStrategy("laptop", []);
  expect(strategy.refinedQuery).toBe("laptop"); // ✓ 有效的 JSON
});

test("nextRequestError(400) — single request fails, next succeeds", async () => {
  const mock = new LLMock({ port: 0 });
  mock.onMessage(/.*/, { content: validJson });

  // 一次性：只有下一個請求回傳 400
  mock.nextRequestError(400, { message: "Bad Request" });

  await mock.start();
  // ... 環境設定 ...

  const reasoner = new Reasoner("gpt-4o-mini");

  // 第一次呼叫：400
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  // 第二次呼叫：一次性錯誤已用掉 → 正常回應
  const strategy = await reasoner.buildSearchStrategy("laptop", []);
  expect(strategy.refinedQuery).toBe("laptop"); // ✓
});