使用LangChain AI Agent查询pandas DataFrame时的结果不一致问题排查

核心解决方案 #

1. 构建标准化映射预处理层 #

在Agent生成pandas查询前，加入一个统一的国家标识标准化步骤，避免直接使用用户输入的原始值：

• 维护一个覆盖所有常见别名的映射字典，包含缩写、全称、大小写变体等：

  country_mapping = {
      "pr": "PRI", "mx": "MEX",
      "mexico": "MEX", "puerto rico": "PRI",
      "墨西哥": "MEX", "波多黎各": "PRI"
      # 按需扩展所有需要适配的别名
  }
  

• 编写预处理函数，自动替换用户输入中的非标准标识：

  def normalize_country_input(query):
      lower_query = query.lower()
      for alias, standard in country_mapping.items():
          lower_query = lower_query.replace(alias, standard)
      return lower_query
  

• 将该函数集成到Agent的输入解析流程中，确保生成查询前先完成标准化转换。

2. 优化预提示词的指令约束 #

强化提示词中关于国家标识处理的明确要求，避免Agent自行推断：

处理国家相关查询时，必须先将用户输入的国家名称/代码与提供的标准化映射表匹配，使用映射后的标准值生成pandas查询，禁止直接使用用户输入的原始标识。若未找到匹配项，需告知用户无法识别该国家，而非直接执行无效查询。

可以在提示词中直接嵌入映射表的核心内容，或者明确告知Agent可以调用专门的映射工具获取标准值。

3. 增强空结果的重试逻辑 #

给Agent添加错误处理工具，当查询返回空结果时自动触发二次验证：

• 编写工具函数，当检测到空结果时，自动尝试用映射表中的别名重新查询：

  def retry_with_mapped_country(df, query_col, original_value):
      for alias, standard in country_mapping.items():
          if original_value.lower() == alias:
              result = df[df[query_col] == standard]
              if not result.empty:
                  return result
      return None
  

• 在Agent的工具列表中添加该函数，并在提示词中明确：当查询返回空结果时，必须调用此工具进行重试。

4. 引入轻量实体识别辅助 #

使用小型NLP模型自动提取用户输入中的国家实体，减少手动映射的遗漏：

• 借助spaCy的轻量模型识别国家/地区实体：

  import spacy
  nlp = spacy.load("en_core_web_sm")
  
  def extract_country_entity(query):
      doc = nlp(query)
      for ent in doc.ents:
          if ent.label_ == "GPE":
              return ent.text.lower()
      return None
  

• 将提取到的实体传入映射函数，转换为标准标识后再生成查询。

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