Python调用国内常用AI API的3大实战案例详解

pip install openai pandas python-dotenv

# .env 文件内容
API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

import os
from dotenv import load_dotenv
# 加载 .env 文件中的环境变量
load_dotenv()
api_key = os.getenv("API_KEY")
print("API Key 加载成功！" if api_key else "请检查 .env 文件")

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()

# 初始化客户端（兼容 OpenAI 格式的接口）
client = OpenAI(
    api_key=os.getenv("API_KEY"),
    base_url="https://api.siliconflow.cn/v1"  # 硅基流动的接口地址
)


def summarize_text(text, model="deepseek-ai/DeepSeek-V2.5"):
    """
    用 AI 总结长文本

    参数：
        text: 要总结的原始文本
        model: 使用的模型
    返回：
        AI 生成的总结
    """

    prompt = f"""请帮我总结以下文章的核心内容，要求：
1. 提取 3-5 个关键要点
2. 每个要点用一句话概括
3. 最后给出一个简短的总结（50字以内）

文章内容：
{text}
"""

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "user", "content": prompt}
        ],
        temperature=0.3,  # 温度低一些，结果更稳定
        max_tokens=500    # 限制输出长度
    )

    return response.choices[0].message.content


# 使用示例
if __name__ == "__main__":
    article = """
    2024年是AI大模型爆发的一年。从ChatGPT到Claude，从GPT-4到DeepSeek，
    大模型技术以惊人的速度渗透到各行各业。

    在编程领域，AI辅助编程工具如Cursor、GitHub Copilot正在改变开发者的工作方式。
    调查数据显示，使用AI辅助编程的开发者，平均编码效率提升了40%以上。

    展望2025年，AI Agent（AI代理）成为新的技术热点。
    与简单的问答不同，AI Agent能够自主规划任务、调用工具、反思结果，
    初步具备了"自动化执行"的能力。
    """

    summary = summarize_text(article)
    print("=== 文章总结 ===")
    print(summary)

def batch_summarize(articles, delay=1):
    """
    批量总结多篇文章

    参数：
        articles: 文章列表，每项为 {"title": "标题", "content": "内容"}
        delay: 每次请求间隔（秒），避免请求过快
    """
    results = []

    for i, article in enumerate(articles):
        print(f"正在处理第 {i+1}/{len(articles)} 篇: {article['title']}")

        try:
            summary = summarize_text(article['content'])
            results.append({
                "标题": article['title'],
                "总结": summary,
                "状态": "成功"
            })
        except Exception as e:
            results.append({
                "标题": article['title'],
                "总结": "",
                "状态": f"失败: {e}"
            })

        # 控制请求频率
        if i < len(articles) - 1:
            import time
            time.sleep(delay)

    return results


# 使用示例
if __name__ == "__main__":
    articles = [
        {
            "title": "AI编程工具评测",
            "content": "本文深入评测了当前主流的AI编程工具..."
        },
        {
            "title": "Python异步编程详解",
            "content": "Python的asyncio模块为异步编程提供了强大支持..."
        },
        {
            "title": "数据可视化最佳实践",
            "content": "好的数据可视化能让复杂数据一目了然..."
        }
    ]

    results = batch_summarize(articles)

    # 保存结果
    import pandas as pd
    df = pd.DataFrame(results)
    df.to_excel("文章总结报告.xlsx", index=False)
    print("\n批量总结完成，结果已保存到 文章总结报告.xlsx")

def generate_weekly_report(raw_notes, model="deepseek-ai/DeepSeek-V2.5"):
    """
    用 AI 根据原始工作记录生成周报

    参数：
        raw_notes: 原始工作记录（可以是凌乱的文字）
        model: 使用的模型
    返回：
        格式化后的周报
    """

    prompt = f"""你是一个专业的周报助手。请根据以下工作记录，帮我生成一份结构化的周报。

要求格式如下：
## 本周工作总结
（用列表形式，清晰列出完成的工作）

## 关键成果
（用列表形式，列出最重要的 2-3 项成果）

## 遇到的问题
（如果有的话）

## 下周计划
（根据本周工作，合理推测下周计划）

## 本周心得
（一句简短的反思或总结）

---
原始工作记录：
{raw_notes}
"""

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "user", "content": prompt}
        ],
        temperature=0.5,
        max_tokens=1000
    )

    return response.choices[0].message.content


# 使用示例
if __name__ == "__main__":
    # 随便写的工作记录，不需要整理
    raw = """
    周一：开了个项目评审会，讨论了下个版本的需求
    周二：写了两个接口，调试了半天发现有个bug
    周三：修复了周二的bug，测试通过了
    周四：和产品经理对了需求，改了一个小细节
    周五：写了点文档，review了同事的代码

    另外周二那天还顺便帮同事看了一个线上问题
    """

    report = generate_weekly_report(raw)
    print("=== 本周周报 ===")
    print(report)

=== 本周周报 ===
## 本周工作总结
- 参与项目评审会议，讨论下版本需求
- 完成两个接口的开发工作
- 定位并修复了接口 bug，测试全部通过
- 与产品经理确认需求变更并完成调整
- 完成技术文档编写，完成同事代码 review
## 关键成果
- 🔧 解决了接口 bug，保证版本按时提测
- 📝 完成了新功能的接口文档
## 遇到的问题
- 接口联调时发现数据格式不统一，经过排查修复
## 下周计划
- 继续跟进需求变更
- 准备下版本开发工作
## 本周心得
- 做好接口自测可以大大减少联调时间

import requests

def send_to_dingtalk(webhook_url, content):
    """发送消息到钉钉群（需要机器人 webhook）"""

    data = {
        "msgtype": "text",
        "text": {
            "content": f"📊 本周周报\n\n{content}"
        }
    }

    response = requests.post(webhook_url, json=data)
    return response.json()


# 使用示例
# 1. 在钉钉群设置 -> 智能群助手 -> 添加机器人
# 2. 选择「自定义」机器人，复制 Webhook 地址
# 3. 将地址填入下方
webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=xxx"

report = generate_weekly_report(raw)
result = send_to_dingtalk(webhook_url, report)
print("发送成功！" if result.get("errcode") == 0 else f"发送失败: {result}")

import pandas as pd
from openai import OpenAI
import os
from dotenv import load_dotenv
import time

load_dotenv()

client = OpenAI(
    api_key=os.getenv("API_KEY"),
    base_url="https://api.siliconflow.cn/v1"
)


def classify_text(text, categories, model="deepseek-ai/DeepSeek-V2.5"):
    """
    用 AI 对文本进行分类

    参数：
        text: 要分类的文本
        categories: 可选的分类列表
        model: 使用的模型
    返回：
        分类结果
    """

    prompt = f"""请将以下文本分类到给定类别中。

可选类别：{', '.join(categories)}

要求：
1. 只选择一个最合适的类别
2. 如果文本太短无法判断，返回「无法分类」
3. 直接输出类别名称，不要解释

文本：
{text}
"""

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "user", "content": prompt}
        ],
        temperature=0.1,  # 温度低，分类结果更一致
        max_tokens=50
    )

    return response.choices[0].message.content.strip()


def batch_classify_data(input_file, text_column, categories, output_file=None):
    """
    批量分类 Excel 中的文本数据

    参数：
        input_file: 输入 Excel 文件路径
        text_column: 要分类的文本列名
        categories: 分类列表
        output_file: 输出文件路径（默认覆盖原文件）
    """

    # 读取 Excel
    df = pd.read_excel(input_file)
    print(f"读取文件，共 {len(df)} 条数据")

    # 添加分类列
    df['AI分类'] = ''
    df['分类时间'] = ''

    for i, row in df.iterrows():
        text = str(row[text_column])

        try:
            category = classify_text(text, categories)
            df.at[i, 'AI分类'] = category
            print(f"[{i+1}/{len(df)}] ✅ {category}: {text[:20]}...")

        except Exception as e:
            df.at[i, 'AI分类'] = f"分类失败: {e}"
            print(f"[{i+1}/{len(df)}] ❌ 失败: {e}")

        # 添加延迟，避免请求过快
        if i < len(df) - 1:
            time.sleep(0.8)

    # 保存结果
    if output_file is None:
        output_file = input_file

    df.to_excel(output_file, index=False)
    print(f"\n分类完成！结果已保存到: {output_file}")

    return df


# 使用示例
if __name__ == "__main__":
    # 定义分类类别
    categories = ["功能咨询", "Bug反馈", "投诉建议", "商务合作", "无效信息"]

    # 对 Excel 中的「用户反馈」列进行分类
    batch_classify_data(
        input_file="客户反馈表.xlsx",
        text_column="用户反馈",
        categories=categories,
        output_file="客户反馈表_已分类.xlsx"
    )

class AITool:
    """AI 工具封装类"""

    def __init__(self, api_key, base_url="https://api.siliconflow.cn/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def ask(self, question, context="", model="deepseek-ai/DeepSeek-V2.5", temperature=0.7):
        """
        向 AI 提问

        参数：
            question: 你的问题
            context: 背景上下文（可选）
            model: 模型
            temperature: 创造性程度（0-1，越高越有创意）
        返回：
            AI 的回答
        """
        if context:
            prompt = f"背景信息：\n{context}\n\n问题：\n{question}"
        else:
            prompt = question

        response = self.client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature
        )

        return response.choices[0].message.content

    def summarize(self, text, points=3):
        """总结文本，points 指定要点数量"""
        prompt = f"请总结以下内容的核心要点（{points}个）：\n\n{text}"
        return self.ask(prompt, temperature=0.3)

    def translate(self, text, target_lang="中文"):
        """翻译文本"""
        prompt = f"请将以下内容翻译成{target_lang}：\n\n{text}"
        return self.ask(prompt, temperature=0.3)

    def classify(self, text, categories):
        """文本分类"""
        prompt = f"将以下文本分类到 {', '.join(categories)} 中，直接输出类别：\n\n{text}"
        return self.ask(prompt, temperature=0.1)


# 使用示例
if __name__ == "__main__":
    ai = AITool(api_key=os.getenv("API_KEY"))

    # 总结
    summary = ai.summarize("你的长文本...", points=3)
    print(summary)

    # 翻译
    translation = ai.translate("Hello world!", target_lang="中文")
    print(translation)

    # 分类
    category = ai.classify("这个功能怎么用？", ["咨询", "Bug", "其他"])
    print(category)

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_with_retry(client, messages):
    """带重试的 API 调用"""
    return client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V2.5",
        messages=messages
    )