如何为LLM长链GPT Agent创建带预定义列的CSV分析函数工具

背景 #

我要创建一个GPT Agent，它需要从3个预定义CSV文件中获取数据，因此得为每个CSV文件创建对应的函数工具。

核心需求 #

这个GPT Agent必须清楚每个CSV的用途、包含的列以及各列能提供的数据细节——在工具定义里明确CSV结构和列信息，能大幅提升LLM生成结果的效率和准确性。

当前问题 #

我不确定该如何正确定义CSV的结构与列定义，下面是我尝试写的代码：

csv1_inspection_tool = StructuredTool.from_function(
    func=get_first_n_rows,
    name="InspectCSVFile",
    description="Explore the contents and structure of a table document, displaying its column names and the first n rows, with n defaulting to 3.",
)

import pandas as pd

def get_csv_filename(
        filename: str
    ) -> str:
    """Get CSV file name"""
    # Read the CSV file
    csv_file = pd.read_csv(filename)
    
    # Since there's no sheet name, we just return the filename
    return f"The file name of the CSV is '{filename}'"


def get_column_names(filename: str) -> str:
    """Get all column names from a CSV file"""

    df = pd.read_csv(filename)
    column_names = '\n'.join(df.columns.to_list())
    result = f"The File '{filename}' has columns:\n\n{column_names}"
    return result


def get_first_n_rows(
        filename: str,
        n: int = 3
) -> str:
    """Get CSV File First N Lines"""
    result = get_csv_filename(filename) + "\n\n"
    result += get_column_names(filename) + "\n\n"

    df = pd.read_csv(filename)  
    n_lines = '\n'.join(
        df.head(n).to_string(index=False, header=True).split('\n')
    )
    result += f"This file '{filename}' has first {n} lines of sample:\n\n{n_lines}"
    return result

优化方案 #

1. 为每个预定义CSV创建专属工具 #

既然是固定的3个CSV，没必要用通用工具，直接为每个CSV创建专属工具，在描述里明确其业务用途、核心列的含义，让Agent一眼就能区分：

from langchain.tools import StructuredTool

# 假设3个CSV分别是用户数据、订单数据、产品数据
# 用户数据CSV工具
user_csv_tool = StructuredTool.from_function(
    func=lambda n=3: get_first_n_rows("user_data.csv", n),
    name="InspectUserCSV",
    description="用于查看用户数据CSV文件，该文件存储平台注册用户的基础信息，包含列：user_id（用户唯一ID）、username（用户名）、email（邮箱）、register_date（注册日期）、age（年龄）。工具会返回文件名、所有列名及前n行数据，n默认值为3。"
)

# 订单数据CSV工具
order_csv_tool = StructuredTool.from_function(
    func=lambda n=3: get_first_n_rows("order_data.csv", n),
    name="InspectOrderCSV",
    description="用于查看订单数据CSV文件，该文件存储用户的订单交易信息，包含列：order_id（订单唯一ID）、user_id（关联用户ID）、product_id（关联产品ID）、order_date（下单日期）、total_amount（订单总金额）、status（订单状态）。工具会返回文件名、所有列名及前n行数据，n默认值为3。"
)

# 产品数据CSV工具
product_csv_tool = StructuredTool.from_function(
    func=lambda n=3: get_first_n_rows("product_data.csv", n),
    name="InspectProductCSV",
    description="用于查看产品数据CSV文件，该文件存储平台在售产品的信息，包含列：product_id（产品唯一ID）、product_name（产品名称）、category（产品分类）、price（单价）、stock（库存数量）。工具会返回文件名、所有列名及前n行数据，n默认值为3。"
)

2. 增强函数的信息返回能力 #

现有函数只返回列名和样本行，还可以补充列的数据类型、空值情况、示例值含义，让Agent更清楚数据细节：

import pandas as pd

def get_csv_full_info(filename: str, n: int = 3) -> str:
    """获取CSV文件的完整结构与数据细节"""
    df = pd.read_csv(filename)
    
    # 基础文件信息
    info = f"文件名：{filename}\n"
    info += f"总行数：{len(df)}，总列数：{len(df.columns)}\n\n"
    
    # 列详细信息
    info += "列详细信息：\n"
    for col in df.columns:
        dtype = str(df[col].dtype)
        non_null_count = df[col].notna().sum()
        null_count = df[col].isna().sum()
        sample_values = df[col].dropna().unique()[:3] if non_null_count >0 else ["无有效数据"]
        info += f"- 列名：{col}\n"
        info += f"  数据类型：{dtype}\n"
        info += f"  非空行数：{non_null_count}，空值行数：{null_count}\n"
        info += f"  示例值：{', '.join(map(str, sample_values))}\n\n"
    
    # 前n行样本
    info += f"前{n}行样本数据：\n"
    info += df.head(n).to_string(index=False)
    return info

更新工具使用这个增强后的函数：

user_csv_tool = StructuredTool.from_function(
    func=lambda n=3: get_csv_full_info("user_data.csv", n),
    name="InspectUserCSV",
    description="用于查看用户数据CSV文件的完整结构与数据细节，该文件存储平台注册用户的基础信息。工具会返回文件基本信息、各列的数据类型/空值情况/示例值，以及前n行样本数据，n默认值为3。"
)

3. 优化函数的重复读取问题 #

现有代码里get_first_n_rows会重复调用pd.read_csv三次，可优化为只读取一次，提升效率：

def get_first_n_rows(
        filename: str,
        n: int = 3
) -> str:
    """Get CSV File First N Lines"""
    df = pd.read_csv(filename)
    
    result = f"The file name of the CSV is '{filename}'\n\n"
    result += f"The File '{filename}' has columns:\n\n" + '\n'.join(df.columns.to_list()) + "\n\n"
    
    n_lines = '\n'.join(
        df.head(n).to_string(index=False, header=True).split('\n')
    )
    result += f"This file '{filename}' has first {n} lines of sample:\n\n{n_lines}"
    return result

总结 #

通过为每个预定义CSV创建专属工具并明确其业务用途、增强函数返回的列细节信息、优化代码性能，就能让GPT Agent精准理解每个CSV的结构与数据含义，从而更高效准确地完成数据获取任务。

内容的提问来源于stack exchange，提问作者John