Python + DeepSeek API：搭建一个AI自动化工作流（含完整代码）

# .env 文件DEEPSEEK_API_KEY=sk-your-api-key-hereDEEPSEEK_BASE_URL=https://api.deepseek.com

安全提示： API Key属于敏感信息，切勿硬编码在代码中或提交到公开仓库。建议使用 .env 文件管理，并将 .env 加入 .gitignore。

import osimport jsonfrom openai import OpenAIfrom dotenv import load_dotenv
load_dotenv()
client = OpenAI(    api_key=os.getenv("DEEPSEEK_API_KEY"),    base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),)
def call_deepseek(    prompt: str,    system_prompt: str = "你是一个高效的AI助手。",    model: str = "deepseek-chat",    temperature: float = 0.3,    max_tokens: int = 2048,    json_mode: bool = False,) -> str:    """通用DeepSeek API调用封装"""    kwargs = {        "model": model,        "messages": [            {"role": "system", "content": system_prompt},            {"role": "user", "content": prompt},        ],        "temperature": temperature,        "max_tokens": max_tokens,    }    if json_mode:        kwargs["response_format"] = {"type": "json_object"}
    response = client.chat.completions.create(**kwargs)    return response.choices[0].message.content

def generate_daily_report(tasks: list[str], date: str = None) -> str:    """根据任务列表生成工作日报"""    from datetime import datetime    if date is None:        date = datetime.now().strftime("%Y年%m月%d日")
    task_list = "\n".join(f"- {t}" for t in tasks)
    system_prompt = """你是一位项目管理专家，擅长将零散的工作记录整理成结构清晰的日报。日报格式要求：1. 按项目/模块分类2. 每项工作标注完成状态（已完成/进行中/待跟进）3. 末尾添加明日计划（根据今日进展推断）4. 语言简洁专业，避免冗余"""
    prompt = f"""请根据以下工作记录，生成{date}的工作日报：
{task_list}
要求：按项目分类，标注完成状态，补充明日计划。"""
    return call_deepseek(prompt, system_prompt=system_prompt)

# === 使用示例 ===if __name__ == "__main__":    today_tasks = [        "修复了用户登录页面的CSS样式问题",        "完成了订单模块的接口对接，POST /api/orders 已通过联调",        "参加了下午3点的需求评审会，讨论了V2.3版本的搜索优化",        "写了一个Python脚本自动从数据库导出日报数据",        "帮新人配置了开发环境",        "Review了小李的PR #287（支付模块），提了3条建议",    ]
    report = generate_daily_report(today_tasks)    print(report)

2026年5月3日 工作日报
【前端开发】- 用户登录页面CSS样式修复 ✅ 已完成- V2.3搜索优化需求评审 🔄 进行中（已参会讨论）
【后端开发】- 订单模块接口对接（POST /api/orders） ✅ 已完成联调- Python数据导出脚本开发 ✅ 已完成
【团队协作】- 新人开发环境配置 ✅ 已完成- 支付模块代码评审（PR #287） ✅ 已完成，已反馈3条建议
明日计划：1. 跟进V2.3搜索优化的技术方案设计2. 处理PR #287反馈的修改项3. 启动用户反馈系统的前期调研

def classify_emails(emails: list[dict]) -> list[dict]:    """批量邮件智能分类"""    system_prompt = """你是一个邮件分类助手。根据邮件标题和摘要，为每封邮件分配：1. category: 客户咨询 | 技术讨论 | 会议通知 | 内部通知 | 市场推广 | 其他2. priority: 高 | 中 | 低3. action_required: 是否需要回复（true/false）4. summary: 一句话摘要（15字以内）
返回JSON格式。"""
    email_data = json.dumps(        [{"id": i, "subject": e["subject"], "preview": e["preview"]}         for i, e in enumerate(emails)],        ensure_ascii=False, indent=2    )
    prompt = f"请分类以下邮件：\n{email_data}"
    result = call_deepseek(        prompt,        system_prompt=system_prompt,        json_mode=True,        temperature=0.1,  # 分类任务需要确定性输出    )    return json.loads(result)

# === 使用示例 ===if __name__ == "__main__":    test_emails = [        {"subject": "关于Q2产品路线图的讨论", "preview": "张总，附件是Q2产品路线图初版，请过目..."},        {"subject": "[紧急] 线上支付接口返回500错误", "preview": "生产环境支付接口从10:30开始大面积报错..."},        {"subject": "5月团建活动报名", "preview": "各位同事，本月团建定于5月15日..."},        {"subject": "Re: React 19迁移方案", "preview": "我觉得可以先用React 19的beta版本在测试环境验证..."},        {"subject": "【推广】云服务器限时5折", "preview": "尊敬的开发者，我们为您准备了专属优惠..."},        {"subject": "周报提醒：请提交本周工作总结", "preview": "请在本周五18:00前提交周报至OA系统..."},    ]
    results = classify_emails(test_emails)
    for r in results:        print(f"[{r['priority']}优先级] {r['category']}")        print(f"  {r['summary']}")        print(f"  需回复: {'是' if r['action_required'] else '否'}")        print()

def suggest_reply(email: dict, classification: dict) -> str:    """为需要回复的邮件生成回复建议"""    system_prompt = "你是一位专业的商务邮件回复助手，语气礼貌简洁。"
    prompt = f"""邮件主题：{email['subject']}邮件摘要：{email['preview']}分类：{classification['category']}优先级：{classification['priority']}
请生成一封回复草稿，50-100字即可。"""
    return call_deepseek(prompt, system_prompt=system_prompt)

# ❌ 低效：逐条调用for email in emails:    result = call_deepseek(f"分类这封邮件: {email}")
# ✅ 高效：批量处理results = classify_emails(emails)  # 一次调用处理全部

def call_deepseek_stream(prompt: str, system_prompt: str = "你是AI助手。"):    """流式输出，逐字返回"""    stream = client.chat.completions.create(        model="deepseek-chat",        messages=[            {"role": "system", "content": system_prompt},            {"role": "user", "content": prompt},        ],        stream=True,    )    for chunk in stream:        if chunk.choices[0].delta.content:            yield chunk.choices[0].delta.content
# 使用for text in call_deepseek_stream("帮我写一封感谢信"):    print(text, end="", flush=True)

import time
def call_with_retry(prompt: str, max_retries: int = 3, **kwargs) -> str:    """带重试的API调用"""    for attempt in range(max_retries):        try:            return call_deepseek(prompt, **kwargs)        except Exception as e:            if attempt < max_retries - 1:                wait = 2 ** attempt  # 指数退避：1s, 2s, 4s                print(f"第{attempt+1}次失败，{wait}秒后重试: {e}")                time.sleep(wait)            else:                raise