Python + 大模型:从零搭建你的数字员工

# 安装依赖pip install openai pandas imaplib email schedule# 配置大模型API（以OpenAI为例，国内可替换为通义千问/文心一言）import openaiopenai.api_key = "你的API密钥"

import imaplibimport emailfrom email.header import decode_headerdef read_unread_emails():    # 连接邮箱服务器（以QQ邮箱为例）    mail = imaplib.IMAP4_SSL("imap.qq.com")    mail.login("你的邮箱@qq.com", "授权码")    mail.select("INBOX")    # 搜索未读邮件    status, messages = mail.search(None, "UNSEEN")    email_ids = messages[0].split()    emails = []    for e_id in email_ids:        status, msg_data = mail.fetch(e_id, "(RFC822)")        for response_part in msg_data:            if isinstance(response_part, tuple):                msg = email.message_from_bytes(response_part[1])                subject = decode_header(msg["Subject"])[0][0]                if isinstance(subject, bytes):                    subject = subject.decode()                # 获取邮件正文                if msg.is_multipart():                    for part in msg.walk():                        if part.get_content_type() == "text/plain":                            body = part.get_payload(decode=True).decode()                else:                    body = msg.get_payload(decode=True).decode()                emails.append({"subject": subject, "body": body})    mail.close()    mail.logout()    return emails

def generate_reply(email_body):    prompt = f"""    你是一名专业的客户服务专员，请根据以下客户邮件内容生成友好、专业的回复：    客户邮件：{email_body}    回复要求：    1. 先感谢客户的咨询；    2. 针对客户问题给出清晰解答；    3. 邀请客户如有其他问题随时联系。    """    response = openai.ChatCompletion.create(        model="gpt-3.5-turbo",        messages=[{"role": "user", "content": prompt}]    )    return response.choices[0].message.content

import smtplibfrom email.mime.text import MIMETextdef send_reply(to_email, reply_content):    msg = MIMEText(reply_content, "plain", "utf-8")    msg["From"] = "你的邮箱@qq.com"    msg["To"] = to_email    msg["Subject"] = "关于您的咨询回复"    server = smtplib.SMTP_SSL("smtp.qq.com", 465)    server.login("你的邮箱@qq.com", "授权码")    server.sendmail("你的邮箱@qq.com", to_email, msg.as_string())    server.quit()

import scheduleimport timedef daily_email_task():    print("开始处理未读邮件...")    emails = read_unread_emails()    for email in emails:        reply = generate_reply(email["body"])        # 这里需要从邮件中提取收件人邮箱，可扩展代码实现        send_reply("客户邮箱@example.com", reply)    print("邮件处理完成！")# 设置每天上午9点执行任务schedule.every().day.at("09:00").do(daily_email_task)while True:    schedule.run_pending()    time.sleep(60)

from transformers import AutoTokenizer, AutoModelForCausalLMtokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-7B-Chat")model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen-7B-Chat", device_map="auto")def local_generate_reply(email_body):    prompt = f"客户邮件：{email_body}\n请生成回复："    inputs = tokenizer(prompt, return_tensors="pt").to("cuda")    outputs = model.generate(**inputs, max_new_tokens=500)    return tokenizer.decode(outputs[0], skip_special_tokens=True)

import fitz  # PyMuPDFdef extract_invoice_info(pdf_path):    doc = fitz.open(pdf_path)    text = ""    for page in doc:        text += page.get_text()    prompt = f"请从以下发票文本中提取发票号码、金额、开票日期：\n{text}"    # 调用大模型提取信息    response = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}])    return response.choices[0].message.content

import requestsdef send_wechat_message(content):    # 企业微信机器人API    url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的机器人key"    data = {"msgtype": "text", "text": {"content": content}}    requests.post(url, json=data)