AskTable

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

AskTable 团队
AskTable 团队 2026年3月5日

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

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

一、Agent 类型与职责

1. DataNodeAgent：SQL 生成专家

职责：

工具：

2. ChartNodeAgent：可视化专家

职责：

工具：

3. PythonNodeAgent：数据处理专家

职责：

工具：

二、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 类型简单

相关阅读：

技术交流：