0Pricing
AI Agents · 课时

由 Pandas 驱动的数据代理工具

为 read_csv、groupby、merge 和 describe 操作定义工具。

由 Pandas 驱动的数据代理工具 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

用于数据代理的 Pandas 工具

您可以不生成任意代码，而是为数据代理提供一组定义明确的 pandas 工具。每个工具执行一项特定的数据操作，并返回结构化结果。

这种方法比自由格式的代码生成更加安全、可预测——代理会选择并串联工具，而不是从头编写代码。

工具：read_csv

这是任何数据分析会话的入口。read_csv 会将 CSV 文件加载到内存中，并返回其形状和列的描述，让代理了解正在处理的数据。

import pandas as pd
from typing import Optional

# In-memory dataframe store (shared across tool calls in one session)
dataframes = {}

def read_csv(path: str, df_name: str = 'df') -> dict:
    '''Load a CSV file and return schema info.'''
    df = pd.read_csv(path)
    dataframes[df_name] = df
    return {
        'df_name': df_name,
        'rows': len(df),
        'columns': list(df.columns),
        'dtypes': df.dtypes.astype(str).to_dict(),
        'sample': df.head(3).to_dict(orient='records'),
        'null_counts': df.isnull().sum().to_dict()
    }

# Agent usage:
result = read_csv('orders.csv', 'orders')
print(f"Loaded {result['rows']} rows, columns: {result['columns']}")

工具：filter_dataframe

根据条件字符串筛选行。该条件会通过 df.query() 在数据框上求值；df.query() 支持安全的 pandas 表达式子集。

def filter_dataframe(df_name: str, condition: str, result_name: str = None) -> dict:
    '''Filter rows matching condition. Returns filtered dataframe info.'''
    df = dataframes.get(df_name)
    if df is None:
        return {'error': f'Dataframe {df_name!r} not found. Call read_csv first.'}

    try:
        filtered = df.query(condition)
        store_name = result_name or f'{df_name}_filtered'
        dataframes[store_name] = filtered
        return {
            'df_name': store_name,
            'rows_before': len(df),
            'rows_after': len(filtered),
            'sample': filtered.head(3).to_dict(orient='records')
        }
    except Exception as e:
        return {'error': f'Filter error: {e}'}

# Example: filter orders from 2024
result = filter_dataframe('orders', 'order_date >= "2024-01-01"', 'orders_2024')
print(f"Filtered to {result['rows_after']} rows")

工具：groupby_aggregate

按某一列对行进行分组并计算聚合值。这是最常见的数据分析操作之一，例如按类别计算总数、平均值和计数。

def groupby_aggregate(df_name: str, group_column: str,
                      agg_column: str, agg_func: str) -> dict:
    '''Group by column and aggregate. agg_func: sum, mean, count, min, max, median'''
    df = dataframes.get(df_name)
    if df is None:
        return {'error': f'Dataframe {df_name!r} not found'}

    VALID_FUNCS = {'sum', 'mean', 'count', 'min', 'max', 'median', 'std'}
    if agg_func not in VALID_FUNCS:
        return {'error': f'Invalid agg_func. Choose from: {VALID_FUNCS}'}

    try:
        result = df.groupby(group_column)[agg_column].agg(agg_func).reset_index()
        result.columns = [group_column, f'{agg_column}_{agg_func}']
        result = result.sort_values(f'{agg_column}_{agg_func}', ascending=False)
        return {
            'result': result.head(20).to_dict(orient='records'),
            'total_groups': len(result)
        }
    except Exception as e:
        return {'error': str(e)}

# Agent call: total revenue by product category
print(groupby_aggregate('orders', 'category', 'revenue', 'sum'))

工具：merge_dataframes

根据公共键连接两个数据框。这样代理就能合并多个文件中的数据，例如将订单与客户表连接起来，以获取客户姓名。

def merge_dataframes(df1_name: str, df2_name: str,
                     on: str, how: str = 'inner',
                     result_name: str = 'merged') -> dict:
    '''Join two dataframes on a key column.'''
    df1 = dataframes.get(df1_name)
    df2 = dataframes.get(df2_name)
    if df1 is None:
        return {'error': f'{df1_name!r} not found'}
    if df2 is None:
        return {'error': f'{df2_name!r} not found'}

    VALID_HOW = {'inner', 'left', 'right', 'outer'}
    if how not in VALID_HOW:
        return {'error': f'Invalid join type. Choose from: {VALID_HOW}'}

    try:
        merged = df1.merge(df2, on=on, how=how)
        dataframes[result_name] = merged
        return {
            'df_name': result_name,
            'rows': len(merged),
            'columns': list(merged.columns),
            'sample': merged.head(3).to_dict(orient='records')
        }
    except Exception as e:
        return {'error': str(e)}

# Join orders with customers on customer_id
print(merge_dataframes('orders', 'customers', on='customer_id'))

工具：describe_dataframe

返回数据框的统计摘要。这通常是代理加载数据后调用的第一个工具，用于了解分布、范围和潜在的数据质量问题。

def describe_dataframe(df_name: str) -> dict:
    '''Return statistical summary and data quality info.'''
    df = dataframes.get(df_name)
    if df is None:
        return {'error': f'Dataframe {df_name!r} not found'}

    # Numeric stats
    numeric_cols = df.select_dtypes(include='number').columns.tolist()
    stats = {}
    for col in numeric_cols:
        s = df[col].describe()
        stats[col] = {
            'min':    round(float(s['min']), 4),
            'max':    round(float(s['max']), 4),
            'mean':   round(float(s['mean']), 4),
            'median': round(float(df[col].median()), 4),
            'std':    round(float(s['std']), 4),
            'nulls':  int(df[col].isnull().sum())
        }

    # Categorical columns summary
    cat_cols = df.select_dtypes(include='object').columns.tolist()
    cat_stats = {
        col: {
            'unique_values': df[col].nunique(),
            'top_3': df[col].value_counts().head(3).to_dict()
        }
        for col in cat_cols
    }

    return {'numeric': stats, 'categorical': cat_stats, 'total_rows': len(df)}

工具：sort_and_top_n

返回按某一列排序后的前 N 行。适用于“顶级客户”“最佳产品”“最高收入”等类型的问题。

def sort_and_top_n(df_name: str, sort_column: str,
                   n: int = 10, ascending: bool = False) -> dict:
    '''Return top N rows sorted by column.'''
    df = dataframes.get(df_name)
    if df is None:
        return {'error': f'{df_name!r} not found'}
    if sort_column not in df.columns:
        return {'error': f'Column {sort_column!r} not found. Available: {list(df.columns)}'}

    sorted_df = df.sort_values(sort_column, ascending=ascending)
    top = sorted_df.head(n)

    return {
        'rows': top.to_dict(orient='records'),
        'total_rows_in_df': len(df),
        'sort_column': sort_column,
        'sort_order': 'ascending' if ascending else 'descending'
    }

# Top 5 orders by total value
print(sort_and_top_n('orders', 'total_amount', n=5))

工具：pivot_table

创建数据透视表——对两个分类列进行交叉制表，并计算聚合值。非常适合“按地区和产品类别统计收入”这类问题。

def pivot_table(df_name: str, index: str, columns: str,
                values: str, aggfunc: str = 'sum') -> dict:
    '''Create a pivot table.'''
    df = dataframes.get(df_name)
    if df is None:
        return {'error': f'{df_name!r} not found'}

    VALID_FUNCS = {'sum', 'mean', 'count', 'min', 'max'}
    if aggfunc not in VALID_FUNCS:
        return {'error': f'Invalid aggfunc. Choose from: {VALID_FUNCS}'}

    try:
        pivot = df.pivot_table(
            index=index, columns=columns,
            values=values, aggfunc=aggfunc,
            fill_value=0
        )
        # Convert to nested dict for JSON serialization
        result = pivot.round(2).to_dict()
        return {
            'pivot': result,
            'index': index,
            'columns': columns,
            'values': values,
            'aggfunc': aggfunc
        }
    except Exception as e:
        return {'error': str(e)}

工具：calculate_metric

根据现有数据框列计算单项业务指标，例如收入增长率、转化率和流失率。相比让 LLM 编写 pandas 表达式，这种方式更加精准。

def calculate_metric(df_name: str, metric: str, params: dict = None) -> dict:
    '''Calculate a named business metric from the dataframe.'''
    df = dataframes.get(df_name)
    if df is None:
        return {'error': f'{df_name!r} not found'}

    params = params or {}
    try:
        if metric == 'conversion_rate':
            total_col = params.get('total_column', 'total')
            converted_col = params.get('converted_column', 'converted')
            rate = df[converted_col].sum() / df[total_col].sum() * 100
            return {'metric': 'conversion_rate', 'value': round(rate, 2), 'unit': '%'}

        elif metric == 'average_order_value':
            revenue_col = params.get('revenue_column', 'revenue')
            count_col = params.get('count_column', 'order_count')
            aov = df[revenue_col].sum() / df[count_col].sum()
            return {'metric': 'average_order_value', 'value': round(aov, 2)}

        else:
            return {'error': f'Unknown metric: {metric}'}

    except Exception as e:
        return {'error': str(e)}

向 LLM 代理注册工具

使用 OpenAI 的函数调用格式，将所有 pandas 工具注册到 LLM 代理中。代理会根据用户的问题决定调用哪个工具。

tools = [
    {
        'type': 'function',
        'function': {
            'name': 'read_csv',
            'description': 'Load a CSV file for analysis',
            'parameters': {
                'type': 'object',
                'properties': {
                    'path': {'type': 'string', 'description': 'Path to CSV file'},
                    'df_name': {'type': 'string', 'description': 'Name to reference this dataframe'}
                },
                'required': ['path']
            }
        }
    },
    {
        'type': 'function',
        'function': {
            'name': 'groupby_aggregate',
            'description': 'Group data by a column and compute sum, mean, count, min, max, or median',
            'parameters': {
                'type': 'object',
                'properties': {
                    'df_name':      {'type': 'string'},
                    'group_column': {'type': 'string'},
                    'agg_column':   {'type': 'string'},
                    'agg_func':     {'type': 'string', 'enum': ['sum', 'mean', 'count', 'min', 'max', 'median']}
                },
                'required': ['df_name', 'group_column', 'agg_column', 'agg_func']
            }
        }
    }
]

if __name__ == '__main__':
    print('Registered tools:')
    for t in tools:
        fn = t['function']
        print(f"  - {fn['name']}: {fn['description']}")

串联多个工具

复杂问题需要串联多个工具。代理会按顺序调用这些工具，将一个工具的输出作为下一个工具的输入。每次工具调用后，工具结果都会先返回给代理，再进行下一次调用。

import json
import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

TOOL_MAP = {
    'read_csv': read_csv,
    'filter_dataframe': filter_dataframe,
    'groupby_aggregate': groupby_aggregate,
    'describe_dataframe': describe_dataframe,
    'merge_dataframes': merge_dataframes,
    'sort_and_top_n': sort_and_top_n
}

def run_data_agent(question, data_paths):
    messages = [
        {'role': 'system', 'content': f'You are a data analyst. Available data: {data_paths}'},
        {'role': 'user', 'content': question}
    ]

    for _ in range(10):  # max 10 tool calls
        response = client.chat.completions.create(
            model='gpt-4o', messages=messages, tools=tools
        )
        msg = response.choices[0].message

        if not msg.tool_calls:
            return msg.content  # final answer

        messages.append(msg)
        for call in msg.tool_calls:
            fn = TOOL_MAP[call.function.name]
            args = json.loads(call.function.arguments)
            result = fn(**args)
            messages.append({'role': 'tool', 'tool_call_id': call.id,
                             'content': json.dumps(result)})

    return 'Max tool calls reached'

知识检查

与使用代码解释器模式相比，为数据代理提供 pandas 工具的主要优势是什么？

回顾：由 Pandas 驱动的数据代理工具

Pandas 工具为数据代理提供了安全且定义明确的操作：read_csv、filter_dataframe、groupby_aggregate、merge_dataframes、describe_dataframe、sort_and_top_n 和 pivot_table。

每个工具都会验证输入、妥善处理错误，并返回结构化结果。使用 OpenAI 的函数调用格式注册工具后，代理会自动串联工具调用，以回答复杂的多步骤数据问题。

常见问题解答

「由 Pandas 驱动的数据代理工具」课时是免费的吗？

是的 — 「由 Pandas 驱动的数据代理工具」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「由 Pandas 驱动的数据代理工具」这节课中我会学到什么？

为 read_csv、groupby、merge 和 describe 操作定义工具。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「由 Pandas 驱动的数据代理工具」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 用于数据分析的代码解释器模式
  2. 由 Pandas 驱动的数据代理工具
  3. 自动生成图表与可视化内容
  4. 统计摘要代理
← 返回 AI Agents