第 7 课：可插拔的数据源

7.1 你的目标

在不改 Agent 代码的前提下，切换数据源（从演示数据到真实数据库）。

.env 中:
DATA_PROVIDER=demo       → 使用内存演示数据
DATA_PROVIDER=postgres   → 使用 PostgreSQL 数据库

7.2 问题：数据耦合

第 4 课的 _execute_tool：

def _execute_tool(name, args):
    if name == "query_order":
        # 订单数据写死在代码里
        _ORDERS = {
            "ORD-001": {"status": "shipped", ...},
        }
        return _ORDERS.get(args["order_id"])

想换成 PostgreSQL？要改 _execute_tool。想换成从 CSV 读？要改 _execute_tool。想写单元测试？需要准备真实数据。

Agent 的逻辑和数据源耦合在一起了。

7.3 本质问题

不是"代码写得不好"，而是一个架构设计问题：

• Agent 的逻辑（"什么时候应该查订单"）应该稳定
• 数据源（"订单数据从哪来"）应该灵活

稳定的逻辑不应该依赖灵活的实现。

7.4 解法：抽象接口

第一步：定义数据模型

@dataclass(frozen=True)
class Order:
    order_id: str
    status: str
    product: str
    amount: float

第二步：定义抽象接口

class ECommerceDataProvider(ABC):
    @abstractmethod
    def get_order(self, order_id: str) -> Optional[Order]: ...
    @abstractmethod
    def get_shipment(self, order_id: str) -> Optional[Shipment]: ...
    @abstractmethod
    def get_all_knowledge_docs(self) -> list[KnowledgeDoc]: ...

第三步：实现多个版本

class DemoDataProvider(ECommerceDataProvider):
    """内存演示数据"""
    def get_order(self, order_id):
        return _ORDERS.get(order_id)

class PostgresDataProvider(ECommerceDataProvider):
    """PostgreSQL 实现"""
    def __init__(self, conn_string):
        self.conn = psycopg2.connect(conn_string)
    def get_order(self, order_id):
        cursor = self.conn.cursor()
        cursor.execute("SELECT * FROM orders WHERE id = %s", (order_id,))
        row = cursor.fetchone()
        if row:
            return Order(row[0], row[1], row[2], row[3])
        return None

第四步：工厂模式

def get_data_provider():
    provider_type = settings.data_provider
    if provider_type == "demo":
        return DemoDataProvider()
    elif provider_type == "postgres":
        return PostgresDataProvider(settings.database_url)

from app.data.config import get_data_provider

def _execute_tool(name, args):
    provider = get_data_provider()
    
    if name == "query_order":
        order = provider.get_order(args.get("order_id", ""))
        if order:
            return json.dumps({"found": True, "order": asdict(order)})
        return json.dumps({"found": False})

_execute_tool 不再知道数据从哪来。它只知道"调 provider.get_order() 能拿到订单"。

7.6 这和 Web 开发的分层有什么不同？

Web 开发中也常用接口抽象数据层。但 Agent 系统的数据层有两点不同：

1. 数据是为 LLM 服务的。接口设计要考虑 LLM 需要什么数据、以什么格式获取。比如 get_order 返回的 Order 对象会被序列化成 JSON 给 LLM 看，字段名要清晰。

2. 错误处理要友好。如果数据库连接失败，LLM 应该回复"系统暂时无法查询订单"而不是抛出 500 错误。

本课知识点

课后作业

1. 实现一个 CSVDataProvider，从 CSV 文件读取订单数据
2. 实现一个 RedisDataProvider，从 Redis 读取会话历史

面试可能会问

"Data Provider 模式和 Repository 模式有什么区别？"
本质相同，都是隔离数据访问。区别在于命名偏好和抽象粒度。Data Provider 更强调"提供数据"，Repository 更强调"聚合根"。