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

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

const llmock = new LLMock({ port: 0 }); // random port

// route fixtures by the content of the system prompt
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 });

// register static results — the same results come back every time
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...",
    },
  },
  // ... more entries
]);

await vectorMock.start();

const server = createServer((req, res) => {
  // Agent card discovery: 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");

  // real Pinecone gives no such guarantee
  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", // unreachable → falls back
      indexName: "product-knowledge",
    },
    ragTimeoutMs: 500,
  });

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

  // even with RAG down, the agent still works via 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,  # required in 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: construct directly with Part(text="..."), no TextPart wrapper
        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,
            )
        )

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

export class A2AClient {
  // v1.0: agent-card.json (was: 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 plus the A2A-Version header
  async sendMessage(text: string, contextId?: string): Promise<A2AMessage> {
    const rpcRequest = {
      jsonrpc: "2.0",
      id: messageId,
      method: "SendMessage",  // was: "tasks/send"
      params: {
        message: {
          role: "ROLE_USER",  // was: "user"
          parts: [{ text }],  // was: { type: "text", text }
          messageId,
          contextId,
        },
      },
    };

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

    const rpcResponse = await res.json();
    return rpcResponse.result.message;  // was: result (a task object)
  }
}

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

  // register a valid fixture (/.*/ matches every message)
  mock.onMessage(/.*/, {
    content: '{"refinedQuery": "test", "priorities": [], "priceRange": {"min": 0, "max": 100}}',
  });

  // chaos: corrupt the response 100% of the time
  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");

  // the fixture matches, but chaos breaks the response
  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();
  // ... env setup ...

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

  // chaos on — it fails
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  // chaos off — it works
  mock.clearChaos();

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

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

  // one-shot: return 400 for the next request only
  mock.nextRequestError(400, { message: "Bad Request" });

  await mock.start();
  // ... env setup ...

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

  // first call: 400
  await expect(
    reasoner.buildSearchStrategy("laptop", [])
  ).rejects.toThrow();

  // second call: the one-shot is spent → normal response
  const strategy = await reasoner.buildSearchStrategy("laptop", []);
  expect(strategy.refinedQuery).toBe("laptop"); // ✓
});