DoorDash 如何让购物助手不只依赖大模型：Agent、MCP 工具与持久记忆的协同架构

from dataclasses import dataclass, field
from typing import Any, Callable

Tool = Callable[..., dict[str, Any]]


@dataclass
class Memory:
    preferences: dict[str, Any] = field(default_factory=dict)
    sessions: dict[str, list[dict[str, str]]] = field(default_factory=dict)

    def context(self, user_id: str, session_id: str) -> dict[str, Any]:
        return {
            "preferences": self.preferences.get(user_id, {}),
            "recent_turns": self.sessions.get(session_id, [])[-4:],
        }

    def append(self, session_id: str, role: str, content: str) -> None:
        self.sessions.setdefault(session_id, []).append(
            {"role": role, "content": content}
        )


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(self, name: str, tool: Tool) -> None:
        self._tools[name] = tool

    def call(self, name: str, **arguments: Any) -> dict[str, Any]:
        if name not in self._tools:
            raise ValueError(f"Tool not allowed: {name}")
        return self._tools[name](**arguments)


def search_catalog(query: str, vegetarian: bool = False) -> dict[str, Any]:
    # Replace this fixture with live catalog, availability and delivery data.
    products = [
        {"id": "meal-1", "name": "Vegetable curry", "price": 14.0,
         "vegetarian": True, "available": True},
        {"id": "meal-2", "name": "Chicken bowl", "price": 16.0,
         "vegetarian": False, "available": True},
    ]
    matches = [
        item for item in products
        if item["available"] and (not vegetarian or item["vegetarian"])
    ]
    return {"query": query, "items": matches}


def price_cart(item_ids: list[str], budget: float) -> dict[str, Any]:
    prices = {"meal-1": 14.0, "meal-2": 16.0}
    total = sum(prices[item_id] for item_id in item_ids)
    return {"item_ids": item_ids, "total": total, "within_budget": total <= budget}


def classify_intent(message: str) -> str:
    # In production, use constrained model output such as an enum or JSON schema.
    return "recommend" if any(word in message.lower() for word in ["find", "meal", "推荐"]) else "unknown"


def handle(message: str, user_id: str, session_id: str,
           memory: Memory, tools: ToolRegistry) -> dict[str, Any]:
    memory.append(session_id, "user", message)
    context = memory.context(user_id, session_id)
    intent = classify_intent(message)

    if intent != "recommend":
        answer = "I need more detail about what you want to shop for."
        memory.append(session_id, "assistant", answer)
        return {"intent": intent, "answer": answer}

    vegetarian = bool(context["preferences"].get("vegetarian", False))
    results = tools.call("catalog.search", query=message, vegetarian=vegetarian)
    candidate_ids = [item["id"] for item in results["items"][:2]]
    cart = tools.call("cart.price", item_ids=candidate_ids, budget=40.0)

    answer = (
        f"Found {len(candidate_ids)} available option(s). "
        f"Live total: ${cart['total']:.2f}; within budget: {cart['within_budget']}."
    )
    memory.append(session_id, "assistant", answer)
    return {"intent": intent, "answer": answer, "cart_preview": cart}


if __name__ == "__main__":
    memory = Memory(preferences={"user-42": {"vegetarian": True}})
    tools = ToolRegistry()
    tools.register("catalog.search", search_catalog)
    tools.register("cart.price", price_cart)

    result = handle(
        "Find a meal for two under $40",
        user_id="user-42",
        session_id="session-1",
        memory=memory,
        tools=tools,
    )
    print(result)

python shopping_assistant.py