DataMate技术笔记(1)——用户场景和开发过程

核心场景定位 🎯

用户画像：创业公司AI工程师张工
需求：做电商评论情感分析模型
现状困境：
- 商业数据集太贵（几万元起）
- 找到的开源数据集格式不统一，需要大量预处理
- 不知道哪些补充数据对模型有帮助
期望：1小时内找到可直接使用的训练数据

MVP核心价值主张

开发（debug）过程

1. 初始最小可行产品（MVP）概念及基本代理原型设计（谷歌代理开发套件ADK之前）：
• 目标：为数据科学家搭建一个数据共享网络。
• 实现方式：
• provider_agent.py：基本的文件扫描（CSV、JSON格式）和元数据提取。
• requester_agent.py：基于命令行界面（CLI）的代理，用于查询单个数据提供者。
• coordinator_agent.py：一个简单的基于Flask的注册中心，供数据提供者进行自我声明，以及数据请求者进行发现。
• 初始的app.py和web_ui/（包含HTML、CSS、JS文件）用于构建自定义的Web界面。
• 重点：验证基本的点对点（P2P）通信和数据列表功能。

2. Web界面与Flask后端集成——调试阶段：
• 目标：使自定义的Web界面能够与Flask后端（app.py）进行通信，最终实现对代理的编排。
• 挑战与调试：这个阶段主要集中解决前端与后端的通信问题。
• 跨域资源共享（CORS）错误：花费了大量时间解决Web界面（如localhost:8000或8001）与Flask后端（localhost:5003）之间的跨域资源共享问题。
• 错误的API网址：前端的JavaScript使用相对路径进行API调用，导致访问了错误的服务器。
• 端口冲突：Flask应用与现有服务（如占用5000端口的AirTunes）发生冲突。

3. 引入谷歌代理开发套件（ADK）及初始代理结构：
• 目标：集成谷歌代理开发套件（ADK），以构建更复杂的、由大语言模型（LLM）驱动的代理。
• 实现方式：
• 创建了adk_agents/目录结构。
• 使用LlmAgent定义了初始的ADK代理（CoordinatorAgent_ADK、ProviderAgent_ADK、RequesterAgent_ADK）。
• 尝试设置ADK环境（.env文件、谷歌云认证）。

4. ADK设置与配置——调试阶段：
• 挑战与调试：
• Pydantic验证错误：对于LlmAgent，enable_full_history参数无效。
• Vertex AI权限问题：ADK默认使用Vertex AI，导致权限错误。后来通过LiteLlm转而使用免费的Gemini开发者API。
• 依赖缺失：出现ModuleNotFoundError: No module named 'litellm'错误（litellm是LiteLlm的一个依赖项）。
• litellm配置问题：litellm仍然试图使用Vertex AI；通过在模型名称前加上gemini/前缀来解决此问题。
• Python导入错误：出现一系列复杂的导入错误（如Module adk_agents not found、cannot import name 'root_agent'）。这些错误是由最初的litellm问题导致的导入链中断、app.py试图导入已从adk_agents.coordinator_agent中重构出去的名称，以及潜在的.pyc缓存问题共同造成的。

5. 将Flask（app.py）连接到ADK代理——调试阶段：
• 目标：让app.py调用ADK代理来处理API请求。
• 挑战与调试：
• 错误的ADK调用方式：app.py最初尝试在ADK的LlmAgent实例上调用自定义的handle_message()方法，这不是ADK的标准调用方式。
• Pydantic模型与字典问题：ADK的LlmAgent（或其使用的工具）期望接收Pydantic模型实例作为消息，但app.py发送的是原始字典（如AttributeError: 'dict' object has no attribute'model_copy'）。
• ADK的Content类型问题：意识到对于run_async方法，ADK期望消息格式为google.genai.types.Content对象，而不仅仅是直接的自定义Pydantic模型。
• ADK会话管理问题：出现Session not found错误。这是一个多步骤的调试过程：
• 初步修复：尝试在app.py中显式创建会话。
• 问题：Runner仍然在内部查找会话，可能是使用了不同的SessionService实例，或者是由于时间问题。
• 解决方案：确保app.py用于会话创建/查找的InMemorySessionService实例与ADK的Runner使用的是同一个实例。同时修正了get_session API调用的参数。——实际未解决，因为意识到ADK可能有多层封装，与其解决这个问题不如pivot，用agent调用网络请求，而不是网络请求调用agent
• Flask同步与ADK异步问题：Flask路由是同步的，但ADK操作（如会话管理和run_async）是异步的。这需要使用包装器（如get_or_create_session_sync，使用asyncio.new_event_loop().run_until_complete()）来弥合差距。
• Python NameError错误：由于在app.py中位置不正确，get_or_create_session_sync未定义。

6. 完善产品愿景与架构（混合模型）：
• 认识：最初的ADK集成将ADK代理视为独立的网络服务。用户明确了分布式P2P网络的愿景，即每个用户运行自己的代理实例，同时也认识到在每个节点内使用ADK进行复杂的本地处理的价值。
• 向混合架构转变：
• 网络层（P2P）：现有的非ADK代理代码（agents/）为P2P HTTP通信（数据提供者向协调器注册，数据请求者发现数据提供者）提供了基础。这一层处理节点间的通信。
• 本地ADK处理层：在每个用户的机器上，在其本地代理实例内，ADK用于管理一组专门的子代理（如请求者端的SmartDataFinder、DataIntegrator；提供者端的DataProvider、ComplianceAgent、QualityAgent）。
• app.py作为桥梁/本地服务器：每个机器上的Flask app.py有两个作用：
1. 它是一个HTTP服务器，监听来自其他代理的传入P2P网络请求。
2. 它是本地入口点，接收请求（来自本地用户界面或传入的网络请求），并将它们委托给本地ADK代理系统（如AppCoordinator）。
• ADK Web界面集成：决定利用ADK的内置图形用户界面（GUI）进行用户交互，简化前端开发并提供原生的ADK体验。app.py的Flask端点随后成为ADK代理（通过adk web运行）可以调用的“工具”。

7. 聚焦MVP（改造方案）：
• 认识：在MVP阶段重新审视架构，发现存在架构重复（adk_agents/与agents/功能重叠）、界面分离（多个HTML文件）、缺乏智能推荐以及交互复杂等问题。因此提出v2.0改造方案，目标是构建统一架构，实现清晰分层职责和灵活工具调用。
• 架构优化：
• 统一架构设计：将原有分离的adk_agents/和agents/合并为统一的ADK智能体层，移除冗余的旧HTTP服务代码和分离的HTML界面。
• 分层职责：
• 用户交互层(ADK Web UI)：提供统一对话界面，支持模式选择(Provider/Requester)
• 智能体业务层(ADK Agents)：包含主协调器MasterCoordinator和专用协调器(ProviderCoordinator/RequesterCoordinator)
• 网络服务层(Flask Backend)：处理P2P通信和本地数据管理
• 数据存储层：保持现有data_store/结构
• 核心智能体重构：
• 统一协调智能体：创建MasterCoordinator作为入口点，集成模式切换工具：

MasterCoordinator = Agent(
    tools=[
        switch_to_provider_mode,  # 切换到提供者模式
        switch_to_requester_mode, # 切换到请求者模式
        AgentTool(agent=ProviderCoordinator),
        AgentTool(agent=RequesterCoordinator)
    ]
)

• 专业化子智能体：
• ProviderCoordinator：集成数据提供、合规检查和质量评估工具
• RequesterCoordinator：集成智能搜索和数据整合工具
• 用户交互流程优化：
• Provider模式流程：
1. 用户选择Provider模式
2. AI扫描本地数据文件夹并自动评估质量
3. AI生成合规报告和分享建议
4. 用户确认后开始网络分享
• Requester模式流程：
1. 用户描述数据需求
2. AI智能分析和扩展需求
3. 自动搜索网络中的匹配数据
4. AI生成数据整合方案和预处理代码
5. 用户一键获取数据和处理代码
• 技术实现策略：
• 保持现有优势：
• Flask后端：继续处理HTTP API和文件操作
• ADK智能体：专注智能逻辑和用户交互
• 本地数据存储：保持data_store/结构不变
• 简化集成点：

# app.py作为工具服务层
@app.route('/api/network/search')
def network_search():
    # 网络通信逻辑

# ADK智能体调用API作为tools
@tool
async def search_network_data(query: str):
    response = requests.post('/api/network/search', json={'query': query})
    return response.json()

• 部署策略：
• 开发阶段：同时运行python app.py(Flask界面)和adk web(ADK界面)
• 生产阶段：adk web作为统一入口，Flask作为后台服务运行

• Python：核心编程语言。

• Flask：用于app.py的Web框架（最初作为主要后端，现在演变为ADK代理的“工具服务器”）。
• 路由、请求处理、JSON响应。
• Flask-CORS用于处理跨域请求。

• HTML、CSS、JavaScript：用于自定义Web界面（web_ui/）。
• fetch API用于客户端-服务器通信。
• 文档对象模型（DOM）操作，用于显示结果。

• 谷歌代理开发套件（ADK）：用于构建多代理应用程序的框架。
• LlmAgent：由大语言模型驱动的代理的核心类。
• LiteLlm：用于使用Gemini API模型（免费层级）的包装器。
• AgentTool：代理使用工具（其他代理或Python函数）的机制。
• Runner、SessionService、Session、Content、Event：运行代理、管理状态和处理通信的ADK核心组件。
• adk web：ADK的内置图形用户界面，用于与代理交互。

• Gemini API（通过谷歌AI工作室）：为ADK代理提供大语言模型能力（免费层级）。

• requests库：用于进行HTTP调用（如在非ADK设置中数据提供者向协调器的调用，以及在混合设置中ADK代理工具对Flask API端点的调用）。

• pandas库：用于数据提供者代理逻辑中的数据操作（如读取CSV文件）。

• JSON：用于API和代理消息的数据交换格式。

• Pydantic：（ADK隐式使用）用于ADK内的数据验证和设置管理。

1. 迭代开发与调试：这个项目是迭代开发的典型例子。你从一个概念开始，构建基本组件，遇到问题，调试问题，然后完善架构。这种构建-测试-调试-完善的循环是软件工程的基础。错误信息是你最好的帮手——理解它们是关键。

2. 关注点分离：
• 最初，存在一些职责模糊的情况（例如，app.py既试图充当Web服务器，又直接管理复杂的代理逻辑）。
• 向混合模型的演进明确了各部分的角色：
• ADK Web界面：用户交互层。
• ADK代理（adk_agents/）：核心智能、决策制定以及复杂本地任务（数据查找、集成、合规性检查、质量评估）的编排。它们定义了需要做“什么”。
• Flask（app.py）：充当强大的“工具服务器”或“服务层”。它处理原始的HTTP请求（既包括ADK Web界面调用工具时的请求，也包括分布式网络中其他P2P代理的请求）。它执行具体的操作，如文件系统操作或P2P网络广播。它定义了特定的、非人工智能任务的“如何”执行方式。
• 这种分层使系统更具模块化、可测试性和可维护性。

3. 特定框架知识：集成任何框架（如Flask或ADK）都需要了解其特定的API、约定和生命周期。许多错误源于最初没有正确使用ADK的Runner、Session或Content对象。阅读新框架的文档和示例至关重要。

4. 同步与异步：将Flask（默认同步）与ADK（异步）连接起来需要一些变通方法（asyncio.new_event_loop().run_until_complete()）。这是集成不同类型系统时常见的挑战。现代Python框架通常提供更好的原生异步支持。

5. 配置与环境管理：API密钥、项目ID（.env文件）以及启用云服务（Vertex AI）等方面的问题凸显了正确配置和理解应用程序运行环境的重要性。

6. 最小可行产品（MVP）的演进：
• 你对MVP的定义本身就经历了演变。最初，它是一个更宽泛的概念。
• 通过讨论，它变得更加聚焦（面向小型人工智能公司的自然语言处理数据交换），这是一种很好的MVP实践。
• MVP的技术实现也从自定义界面加复杂的app.py转变为利用ADK的界面以及ADK代理调用更简单的Flask“工具服务器”。这展示了如何调整技术解决方案，使其更好地适应问题和可用的工具（ADK的GUI）。

7. 清晰通信流程的重要性：了解组件之间如何相互通信（界面 -> app.py -> ADK代理 -> 工具/Flask API）至关重要。图表和清晰的解释（比如你的智能助手提供的或者我们讨论过的那些）有助于巩固这一点。AppCoordinator.handle_message()与Runner.run_async()之间的模糊性就是一个关键的通信/API使用误解。

8. 聚焦的价值：你意识到最初的ADK集成感觉“脱离实际构想的产品场景”，随后转向更聚焦的自然语言处理数据交换场景，同时为ADK（本地处理）和Flask（网络/工具）明确了更清晰的架构角色，这是产品开发中非常成熟的一步。这关乎找到最简单的方式来交付核心价值。

一、核心架构启示

1. 基准点的哲学转变：
• 技术基准 → 用户基准：从以技术组件(P2P/ADK)为中心转向以用户角色(Provider/Requester)为中心
• 神奇之处：当架构基准从"技术实现"转向"用户心智模型"时，系统复杂性不增反降

2. 康威定律的逆向应用：
• 传统：组织架构决定软件架构
• 本例：通过重构软件架构来重塑用户心智模型和组织协作方式
• 验证：将技术团队(P2P组/ADK组)重组为业务团队(Provider流/Requester流)

3. 复杂性的空间转移：

graph LR
A[技术复杂性] -->|转移| B[业务价值]
C[分布式网络复杂性] -->|封装| D[统一协调器]
E[多界面操作] -->|转化| F[自然语言交互]

复杂性从技术层转移到智能体层，但用户感知的复杂性大幅降低

二、架构选择的核心原则

1. 基准选择公式：

最佳基准 = argmin(用户认知成本 × 技术实现成本)

本例证明：以用户角色为基准时，乘积最小

2. 架构演进的辩证法：

阶段	主要矛盾	解决方案
6	技术整合(P2P+ADK)	分层架构
7	用户体验复杂性	角色驱动

3. 工具与代理的范式转变：

# 旧范式：工具为中心
system = NetworkTool() + ADKTool() + FlaskTool()

# 新范式：代理为中心
system = ProviderAgent(
    tools=[NetworkTool, FlaskTool],
    knowledge=[DomainExpertise]
)

三、更高层面的架构哲学

1. 架构即教育：
• 旧架构迫使用户理解技术概念(P2P/代理/端点)
• 新架构教育用户理解业务概念(提供者/请求者)
• 启示：优秀架构应塑造用户心智模型，而非适应用户习惯

2. 复杂性的时空转换：
• 空间复杂性：分布式网络 → 通过统一协调器化解
• 时间复杂性：多步骤操作 → 通过对话流线性化
• 认知复杂性：技术术语 → 通过角色隐喻转化

3. AI时代的架构新法则：

graph TB
A[用户意图] --> B(语义理解层)
B --> C{角色路由}
C --> D[Provider管道]
C --> E[Requester管道]
D --> F[技术工具集]
E --> F

定律：N层技术架构 → 3层认知架构(意图-角色-工具)

四、普适性架构原则

1. 基准选择矩阵：

基准类型	适用场景	风险
技术组件基准	基础设施层	用户认知负担
用户角色基准	业务应用层	过度抽象风险
数据流基准	流处理系统	流程僵化

2. 角色驱动架构(RDA)原则：
• 每个角色对应一个顶级协调器
• 角色间通过标准化协议通信
• 技术组件作为角色工具存在

3. 复杂性守恒的创造性应用：
• 不减少总复杂性，但改变分布：

旧：用户承担70% + 系统承担30%
新：用户承担30% + AI承担40% + 系统承担30%

五、未来演进方向

1. 自适应基准架构：

class AdaptiveArchitecture:
    def __init__(self):
        self.benchmark_detector = BenchmarkPredictor()

    def route_request(self, user_input):
        benchmark = self.benchmark_detector.predict(user_input)
        if benchmark == "technical":
            return TechFirstFlow()
        elif benchmark == "role":
            return RoleFirstFlow()
        else:
            return DataFlow()

2. 架构基准的元认知：
• 系统实时监控"基准选择效率"指标：

基准效率 = 用户任务完成时间 / 系统资源消耗

• 自动优化基准选择策略

3. 跨角色协同协议：
• 定义Provider-Requester交互协议
• 实现角色间的智能合约：

contract DataExchange {
    function requestData(RequesterID, Requirements)
       returns (DataToken);

    function provideData(ProviderID, DataToken)
       payable returns (ReputationScore);
}

终极启示：架构即翻译

技术术语体系 → 业务概念体系
网络节点 → 数据提供者
API端点 → 数据请求
代理调用 → 专业助手