PandasAI连接大语言模型进行智能数据分析的实现方法

目标与环境准备

PandasAI旨在通过生成式AI简化数据分析流程，支持使用自然语言代替复杂代码完成数据查询、清洗与可视化。为了稳定接入兼容OpenAI API格式的大模型（如DeepSeek），建议优先使用虚拟环境管理依赖，避免全局版本冲突。

第1步：安装基础依赖库

目的：获取运行环境与底层HTTP通信能力。

入口：系统终端或IDE内置控制台。

动作：执行标准包管理命令拉取组件：

pip install pandasai requests langchain langchain-openai

重要参数：此处以PandasAI 2.X稳定版为例。若需测试官方推荐的3.0 Beta版本，建议改用poetry管理依赖以避免运行时兼容问题。

第2步：继承基类并封装HTTP请求接口

目的：打通PandasAI引擎与大模型平台的通信链路。

入口：新建独立模型适配文件。

动作：PandasAI通过调用自定义类的call方法向LLM发起请求。直接使用requests发送POST请求是兼容性最强的底层方案。需严格遵循平台提供的JSON结构（通常兼容OpenAI协议）：

import pandas as pd
from pandasai import SmartDataframe
from pandasai.llm.base import LLM
import requests

class DeepSeekLLM(LLM):
    def __init__(self, api_url:str, api_key: str, model: str):
        super().__init__()
        self.api_url = api_url
        self.api_key = api_key
        self.model = model        

    def call(self, instruction, context: dict = None, **kwargs) -> str:  
        if not isinstance(instruction, str):
            instruction = str(instruction)

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        messages = [
            {"role": "system", "content": "You are a helpful AI assistant for data analysis."},
            {"role": "user", "content": instruction}
        ]

        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.0
        }

        response = requests.post(self.api_url, headers=headers, json=payload)
        if response.status_code != 200:
            raise Exception(f"LLM API Error: {response.status_code}, {response.text}")

        return response.json()["choices"][0]["message"]["content"]

    @property
    def type(self):
        return "deepseek-custom"

第3步：加载数据并执行自然语言查询

目的：验证配置是否成功，获取结构化查询结果。

入口：主业务脚本。

动作：实例化自定义LLM，传入兼容端点，绑定至SmartDataframe后即可直接提问：

sales_by_country = pd.DataFrame({
    "country": ["United States", "United Kingdom", "France", "Germany", "Italy", "Spain", "Canada", "Australia", "Japan", "China"],
    "sales": [5000, 3200, 2900, 4100, 2300, 2100, 2500, 2600, 4500, 7000]
})

llm = DeepSeekLLM(
    api_url="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
    api_key="sk-xxx", 
    model="deepseek-r1"
)

df = SmartDataframe(sales_by_country, config={"llm": llm})

result = df.chat("列出销售额最高的3个国家。")
print(result)

完成状态：控制台正常输出包含国家名称与对应销售额的格式化片段即代表连接成功。原始运行结果如下所示：

country  sales
0          China   7000
0          China   7000
1  United States   5000
2          Japan   4500

进阶应用：借助LangChain实现批量处理

在真实业务场景中，单轮查询难以满足吞吐量要求。LangChain提供了完善的抽象层与并发控制机制，能够显著提升批量任务的处理速度。其内部会自动处理队列调度，即使底层模型未提供原生批接口也能平滑过渡。

多标签分类批量调用示例

以下为基于LangChain封装的多标签打标函数，可直接替换业务逻辑：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm_client = ChatOpenAI(
    temperature=0.0,
    model_name="deepseek-r1",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="sk-xxx"
)

def llm_labeling(texts: list[str]) -> list[list[str]]:
    system_prompt = """你是一个多标签分类助手，请从以下标签中选出适用于文本的所有标签（可以多选）：
正面, 负面, 价格问题, 物流问题, 推荐, 外观设计, 用户体验

只输出标签，多个标签之间用英文逗号分隔。如果无法判断，则返回空字符串。
"""
    batch_inputs = [system_prompt + '\n' + t for t in texts] 
    res = llm_client.batch(batch_inputs)
    total_class = {"正面", "负面", "价格问题", "物流问题", "推荐", "外观设计", "用户体验"}
    results = []
    
    for item in res:
        content = item.content.strip()
        if not content:
            results.append([])
            continue
        tags = [t.strip() for t in content.split(',')]
        tags = [t for t in tags if t in total_class]
        results.append(tags)
    return results

texts = [
    "这个产品非常好用，值得推荐",
    "物流速度太慢了，体验不好",
    "外观漂亮，使用方便，就是价格稍贵"
]
print(llm_labeling(texts))

执行该脚本后，终端将按顺序打印每条文本对应的清洗后标签数组。对于需要记忆上下文的场景，大模型服务通常还提供多轮对话与会话保持功能；若追求更低延迟，可启用流式输出实时逐字渲染回答内容。

简要总结

通过继承PandasAI的LLM基类并重写call方法，可以灵活接入任意支持RESTful API的大语言模型，彻底告别硬编码查询语句。结合LangChain的批量调用能力，能够进一步释放自动化数据处理的生产力。以上就是PandasAI连接大语言模型进行智能数据分析的详细内容，更多关于生成式AI落地企业级数据管道的资料请关注本站其它相关文章！