部署在Azure FastAPI的Pandas DataFrame Agent生成文件如何获取？

背景说明 #

你在Azure FastAPI应用中部署的create_pandas_dataframe_agent（配置AgentType.OPENAI_FUNCTIONS）在处理大数据量时会提议生成文件，但默认逻辑不会主动保存文件到可访问路径。你的临时方案是提取查询重跑，以下是更直接的优化方法：

两种直接获取生成文件的方法 #

方法1：自定义工具替换默认文件生成逻辑 #

默认Agent的文件生成是临时内置逻辑，你可以自定义工具函数，让Agent调用它来保存文件并返回可访问路径（本地或Azure存储地址）：

from langchain.tools import tool
import pandas as pd
import os
from uuid import uuid4

# 自定义CSV保存工具
@tool
def save_dataframe_to_csv(df: pd.DataFrame, filename: str = None) -> str:
    """将DataFrame保存为CSV，返回文件访问路径"""
    filename = filename or f"query_result_{uuid4().hex}.csv"
    # 优先用Azure Blob存储（持久化），示例用临时目录演示
    save_path = os.path.join("/tmp", filename)
    df.to_csv(save_path, index=False)
    
    # 若用Azure Blob，替换为上传逻辑并返回SAS URL
    # blob_client = your_blob_service_client.get_blob_client(container="data", blob=filename)
    # blob_client.upload_from_path(save_path)
    # return blob_client.generate_shared_access_signature(expiry=datetime.utcnow() + timedelta(hours=1))
    
    return save_path

# 创建Agent时注入自定义工具
agent = create_pandas_dataframe_agent(
    llm,
    df,
    verbose=True,
    prefix=prefix,
    agent_type=AgentType.OPENAI_FUNCTIONS,
    agent_executor_kwargs={"memory": conversation_memory},
    return_intermediate_steps=True,
    allow_dangerous_code=True,
    tools=[save_dataframe_to_csv]  # 添加自定义工具
)

调用Agent后，从返回结果中提取工具返回的路径/URL，直接对接FastAPI的下载接口。

方法2：从中间步骤提取已计算的DataFrame #

利用已开启的return_intermediate_steps=True，直接提取Agent执行后的DataFrame结果，无需重跑查询：

# 调用Agent
response = agent.invoke({"input": "你的查询请求"})

# 遍历中间步骤，提取查询结果DataFrame
generated_df = None
for step in response["intermediate_steps"]:
    # 匹配Python代码执行步骤（对应Agent的查询操作）
    if hasattr(step[0], "tool") and step[0].tool == "python_repl_ast":
        result = step[1]
        if isinstance(result, pd.DataFrame):
            generated_df = result
            break

# 保存为CSV文件
if generated_df is not None:
    save_path = f"/tmp/result_{uuid4().hex}.csv"
    generated_df.to_csv(save_path, index=False)
    # 将路径传给FastAPI下载接口

部署注意事项 #

• Azure App Service的/tmp目录是临时存储，实例重启后文件会丢失，建议将文件上传到Azure Blob Storage，通过SAS URL提供持久化下载。
• FastAPI下载接口示例：

from fastapi import FastAPI
from fastapi.responses import FileResponse
import os

app = FastAPI()

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = os.path.join("/tmp", filename)
    return FileResponse(file_path, media_type="text/csv", filename=filename)

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