如何优化基于OpenAI Azure与LangChain的Pandas Agent性能？

问题背景 #

基于Azure OpenAI与LangChain实现的Pandas Agent处理数据集查询时，存在明显性能延迟：简单查询需8-12秒完成，且有时计算结果已提前生成，但整体流程仍耗时较久。

当前实现代码 #

import pandas as pd
from time import time
from openai_tools import create_pandas_dataframe_agent

# 数据加载
csv_dir_path = 'https://raw.githubusercontent.com/datasciencedojo/datasets/master/titanic.csv'
df = pd.read_csv(csv_dir_path)

# 创建Pandas Agent
agent_executor = create_pandas_dataframe_agent(
    model,
    df,
    verbose=True,
    agent_type="openai-tools", 
    max_iterations=5
)

def agent_query(question):
    time_start = time()
    res = agent_executor.invoke({"input": question})
    time_finish = time()
    print("Query execution time:", time_finish - time_start, "seconds")
    return res['output']

# 示例查询
question = 'whats the percentage of male survivors'
agent_query(question)

已尝试方案 #

• 调整max_iterations参数，响应时间无明显改善
• 初步排查显示延迟可能源于数据加载或模型调用环节

优化建议与瓶颈定位方法 #

1. 数据加载环节优化 #

• 本地缓存数据集：将远程CSV文件下载到本地，直接读取本地文件，避免每次查询都发起网络请求带来的延迟
• 全局复用DataFrame：确保DataFrame仅在初始化阶段加载一次，避免重复读取和解析数据的开销
• 预做数据预处理：对数据集提前进行类型优化（如将Sex列转为category类型）、清理无效值，提升后续查询计算的速度

2. 模型与Agent配置优化 #

• 选择低延迟模型：更换为Azure OpenAI中延迟更低的模型部署（如gpt-3.5-turbo系列的轻量实例），对比不同模型的响应速度
• 精简Agent交互逻辑：
  • 给Agent添加明确指令，比如“直接生成最优Python代码完成计算，无需多余思考步骤”，减少模型的迭代次数
  • 尝试切换agent_type为zero-shot-react-description，部分场景下该类型的Agent迭代步骤更少，响应更快
• 关闭冗余日志：将verbose设为False，减少日志输出的IO耗时

3. 精准定位性能瓶颈 #

• 拆分计时统计：分别统计数据加载、Agent初始化、模型调用、代码执行各环节的耗时，明确延迟来源：

  # 拆分计时示例
  import pandas as pd
  from time import time
  from openai_tools import create_pandas_dataframe_agent
  
  # 统计数据加载耗时
  time_load_start = time()
  csv_dir_path = 'titanic.csv'  # 本地文件路径
  df = pd.read_csv(csv_dir_path)
  print("Data load time:", time() - time_load_start, "seconds")
  
  # 统计Agent初始化耗时
  time_agent_start = time()
  agent_executor = create_pandas_dataframe_agent(
      model,
      df,
      verbose=False,
      agent_type="openai-tools", 
      max_iterations=5
  )
  print("Agent init time:", time() - time_agent_start, "seconds")
  
  # 统计查询调用耗时
  def agent_query(question):
      time_invoke_start = time()
      res = agent_executor.invoke({"input": question})
      print("Invoke total time:", time() - time_invoke_start, "seconds")
      return res['output']
  
  question = 'whats the percentage of male survivors'
  agent_query(question)
  

• 使用LangChain回调函数：自定义CallbackHandler记录每个步骤的具体耗时（如模型调用、Python代码执行的时间），精准定位慢环节

4. 其他优化点 #

• 复用Agent实例：全局维护一个Agent执行器实例，避免每次查询都重新初始化Agent的开销
• 启用流式响应：如果业务场景允许，开启流式输出，让用户更早看到结果，优化感知体验

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