AskTable Canvas Agent 分工协作：多 Agent 系统的设计与实现

在 AskTable Canvas 中，不同类型的节点需要不同的 Agent 来处理：Data 节点生成 SQL、Chart 节点生成图表配置、Python 节点生成数据处理代码。如何设计一个灵活、可扩展的多 Agent 系统？

本文将深入剖析 AskTable Canvas Agent 的分工协作机制。

---

一、Agent 类型与职责

1. DataNodeAgent：SQL 生成专家

职责：

• •理解用户问题
• •检索相关元数据（Schema Linking）
• •生成 SQL 查询
• •执行并返回结果

工具：

• •schema_linker：元数据检索
• •sql_executor：SQL 执行
• •sql_guard：权限控制

2. ChartNodeAgent：可视化专家

职责：

• •理解数据特征
• •选择合适的图表类型
• •生成图表配置（ECharts）
• •优化图表样式

工具：

• •chart_recommender：图表推荐
• •echarts_generator：配置生成

3. PythonNodeAgent：数据处理专家

职责：

• •理解数据处理需求
• •生成 Python 代码
• •执行并返回结果

工具：

• •python_executor：代码执行
• •dataframe_loader：加载父节点数据

---

二、Agent 基类设计

class BaseAgent(ABC):
    """Agent 基类"""

    def __init__(self, node: NodeModel):
        self.node = node
        self.messages: list[dict] = []
        self.result: dict = {}

    @abstractmethod
    async def run(self, message: str) -> AsyncGenerator[dict, None]:
        """流式执行，返回事件流"""
        pass

    @abstractmethod
    def get_result(self) -> dict:
        """获取最终结果"""
        pass

    def get_messages(self) -> list[dict]:
        """获取对话历史"""
        return self.messages

---

三、DataNodeAgent 实现

class DataNodeAgent(BaseAgent):
    """Data 节点 Agent"""

    def __init__(
        self,
        node: NodeModel,
        datasource: DataSourceAdmin,
        reference_context: dict | None = None,
    ):
        super().__init__(node)
        self.datasource = datasource
        self.reference_context = reference_context

    async def run(self, message: str) -> AsyncGenerator[dict, None]:
        """流式执行"""
        # 1. Schema Linking
        yield {"type": "status", "data": "检索元数据..."}
        meta_context = await self.schema_linker.link(message)

        # 2. 生成 SQL
        yield {"type": "status", "data": "生成 SQL..."}
        sql = await self.generate_sql(message, meta_context)
        yield {"type": "sql", "data": sql}

        # 3. 执行 SQL
        yield {"type": "status", "data": "执行查询..."}
        df = await self.datasource.execute_sql(sql)
        yield {"type": "dataframe", "data": df.to_dict()}

        # 4. 生成解释
        yield {"type": "status", "data": "生成解释..."}
        explanation = await self.generate_explanation(message, df)
        yield {"type": "explanation", "data": explanation}

        # 5. 保存结果
        self.result = {
            "sql": sql,
            "dataframe": df,
            "explanation": explanation,
        }

---

四、ChartNodeAgent 实现

class ChartNodeAgent(BaseAgent):
    """Chart 节点 Agent"""

    def __init__(
        self,
        node: NodeModel,
        parent_contexts: list[dict],
    ):
        super().__init__(node)
        self.parent_contexts = parent_contexts

    async def run(self, message: str) -> AsyncGenerator[dict, None]:
        """流式执行"""
        # 1. 分析数据
        yield {"type": "status", "data": "分析数据..."}
        df = self.parent_contexts[0]["dataframe"]

        # 2. 推荐图表类型
        yield {"type": "status", "data": "推荐图表..."}
        chart_type = await self.recommend_chart_type(df, message)
        yield {"type": "chart_type", "data": chart_type}

        # 3. 生成图表配置
        yield {"type": "status", "data": "生成配置..."}
        config = await self.generate_chart_config(df, chart_type, message)
        yield {"type": "config", "data": config}

        # 4. 保存结果
        self.result = {
            "chart_type": chart_type,
            "config": config,
        }

---

五、工具调用机制

class ToolRegistry:
    """工具注册表"""

    def __init__(self):
        self.tools: dict[str, Callable] = {}

    def register(self, name: str, func: Callable):
        """注册工具"""
        self.tools[name] = func

    async def call(self, name: str, **kwargs) -> Any:
        """调用工具"""
        if name not in self.tools:
            raise ValueError(f"Tool {name} not found")
        return await self.tools[name](**kwargs)

# 注册工具
registry = ToolRegistry()
registry.register("execute_sql", execute_sql)
registry.register("generate_chart", generate_chart)

# Agent 调用工具
result = await registry.call("execute_sql", sql=sql, datasource=ds)

---

六、总结

AskTable Canvas Agent 系统通过清晰的职责划分和灵活的工具调用机制，实现了高效的多 Agent 协作：

✅ 职责清晰：每个 Agent 专注于特定任务 ✅ 工具复用：通过工具注册表共享能力 ✅ 流式输出：实时反馈执行进度 ✅ 易于扩展：新增 Agent 类型简单

---

相关阅读：

• •AskTable Canvas 流式执行

技术交流：

• •官网: asktable.com