深度学习人脸检测与识别系统实战解析
2026/7/25 11:26:03
影刀RPA 调用大模型API实战：让自动化流程拥有AI大脑
作者：林焱
什么情况用
在RPA自动化的过程中，你经常会遇到一些「规则搞不定」的情况：
- 抓取了一批商品评论，需要判断每条是好评还是差评
- 爬到了几十篇文章标题，想自动归纳出核心观点
- 收到客户邮件，需要判断紧急程度并分类转发
- 批量生成商品描述、邮件回复、会议纪要等文本内容
传统RPA做这些事情基本没戏——它不是靠规则能解决的。但把这些任务交给大模型（GPT/Claude/文心一言/通义千问），配合影刀的Python节点调用API，你的自动化流程就像突然长出了大脑。
核心场景：需要判断、理解、归纳、生成文本内容的自动化任务。
怎么做
第一步：搞到大模型的API Key
不管你用的是哪家大模型，都需要先拿到API Key。以国内最常用的几个为例：
通义千问（阿里云）：
- 打开阿里云百炼平台（bailian.console.aliyun.com）
- 创建一个应用，获取API Key
- 模型名用
qwen-turbo（便宜）或qwen-plus（效果好）
拼多多店群自动化上架方案
DeepSeek：
5. 访问 platform.deepseek.com
6. 注册后创建API Key
7. 价格极低，效果很好，推荐新手用
费用方面不用太担心——调用一次通常几分钱，处理一千条评论也就几块钱。
第二步：在影刀里封装调用函数
影刀的Python节点支持直接发HTTP请求，我们封装一个通用的大模型调用函数。
importrequestsimportjsonclassAICaller:""" 通用大模型调用类 支持OpenAI兼容的API格式（通义千问/DeepSeek/GLM等都兼容） """def__init__(self,api_key,base_url,model="qwen-turbo"):self.api_key=api_key self.base_url=base_url.rstrip("/")self.model=modeldefchat(self,system_prompt,user_message,temperature=0.7):""" 单轮对话：发一个system prompt和user message，拿回结果 """url=f"{self.base_url}/chat/completions"headers={"Authorization":f"Bearer{self.api_key}","Content-Type":"application/json"}payload={"model":self.model,"temperature":temperature,"messages":[{"role":"system","content":system_prompt},{"role":"user","content":user_message}]}try:resp=requests.post(url,headers=headers,json=payload,timeout=60)resp.raise_for_status()result=resp.json()returnresult["choices"][0]["message"]["content"].strip()exceptrequests.exceptions.Timeout:return"[错误] API调用超时"exceptExceptionase:returnf"[错误]{str(e)}"defclassify(self,text,categories,system_prompt=""):""" 分类任务：给一段文本，返回它属于哪个类别 自动构造prompt，让模型返回JSON格式的分类结果 """ifnotsystem_prompt:system_prompt="你是一个文本分类专家，请严格按JSON格式返回分类结果。"cat_list="、".join(categories)user_msg=f"""请将以下文本分类到这些类别之一：{cat_list}返回格式（只返回JSON，不要有其他内容）： {{"category": "分类结果", "confidence": "高/中/低"}} 待分类文本：{text}"""result=self.chat(system_prompt,user_msg,temperature=0.1)returnresult# ===== 初始化（填入你自己的API Key） =====ai=AICaller(api_key="sk-your-api-key-here",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",model="qwen-turbo")第三步：实战——商品评论情感分析
这是最常见的需求。拿到一堆评论，自动标注好差评。
defanalyze_review_sentiment(ai,review_text):"""分析单条评论的情感"""system_prompt="""你是一个电商评论分析专家。 请分析用户评论的情感倾向，返回JSON格式： {"sentiment": "正面/负面/中性", "reason": "简短理由"} 注意：如果评论包含明显的不满、质量问题、物流投诉，判定为负面。"""result=ai.chat(system_prompt,review_text,temperature=0.1)# 解析JSON结果，处理可能的格式异常try:# 大模型有时候会在JSON外面包markdown代码块if"```"inresult:result=result.split("```")[1]ifresult.startswith("json"):result=result[4:]result=result.strip()returnjson.loads(result)exceptjson.JSONDecodeError:return{"sentiment":"无法判断","reason":f"解析失败:{result[:50]}"}# 示例调用reviews=["物流太慢了，等了5天才到，包装也破了","质量很好，比想象中还要好，下次还来","还行吧，无功无过","客服态度极差，问题拖了三天没解决"]forreviewinreviews:result=analyze_review_sentiment(ai,review)print(f"评论：{review[:30]}...")print(f"情感：{result['sentiment']}")print(f"理由：{result['reason']}")print("---")第四步：实战——智能邮件分类与路由
收到的邮件五花八门，让大模型判断该转给谁。
defclassify_email(ai,email_subject,email_body):"""分类邮件并决定处理方式"""system_prompt="""你是邮件管理助手。请根据邮件内容判断其紧急程度和归属部门。 返回JSON格式： { "department": "销售/技术/财务/客服/行政/其他", "urgency": "紧急/普通/低优先级", "summary": "一句话概括", "suggested_action": "建议的处理动作" }"""user_msg=f"邮件标题：{email_subject}\n\n邮件正文：\n{email_body}"result=ai.chat(system_prompt,user_msg,temperature=0.1)try:if"```"inresult:result=result.split("```")[1]ifresult.startswith("json"):result=result[4:]result=result.strip()returnjson.loads(result)except:return{"department":"其他","urgency":"普通","summary":"解析失败"}# 示例email_result=classify_email(ai,"紧急：服务器502错误，客户无法下单","刚才收到客户投诉，官网下单页面一直报502错误，已经持续半小时了，请尽快处理！")print(json.dumps(email_result,ensure_ascii=False,indent=2))第五步：实战——批量生成SEO标题和描述
电商运营每天要写几百个商品标题，让大模型批量生成。
defgenerate_seo_titles(ai,product_info,count=5):"""根据产品信息生成多个SEO友好的标题"""system_prompt="""你是电商SEO专家。请根据产品信息生成吸引人的商品标题。 要求： 1. 包含核心关键词 2. 长度15-30字 3. 突出卖点和优惠 4. 生成指定数量，用换行分隔"""user_msg=f"""产品信息：{product_info}请生成{count}个SEO友好的商品标题。"""result=ai.chat(system_prompt,user_msg,temperature=0.8)titles=[t.strip("- ").strip()fortinresult.split("\n")ift.strip()]returntitles[:count]# 示例titles=generate_seo_titles(ai,"女士连衣裙，雪纺面料，夏款，收腰设计，限时优惠价199元")fori,tinenumerate(titles,1):print(f"{i}.{t}")第六步：在影刀流程中串联
把上面的Python代码放到影刀的Python节点里，输入来自前面的网页抓取节点，输出传给Excel写入节点。
影刀流程图： ┌──────────────┐ │ 网页抓取节点 │ → 获取商品评论列表 └──────┬───────┘ ↓ ┌──────────────┐ │ Python节点① │ → 循环调用AI判断情感 └──────┬───────┘ ↓ ┌──────────────┐ │ 条件分支节点 │ → 负面评论 → 企微告警 └──────┬───────┘ 正面评论 → 写入Excel ↓ ┌──────────────┐ │ Excel写入节点 │ → 汇总到情感分析报告 └──────────────┘Python节点里接收影刀变量的写法：
# 影刀传过来的变量通过全局变量获取# 假设影刀里设置了变量 review_listimportjson reviews=json.loads(review_list)# 影刀传来的JSON字符串results=[]forreviewinreviews:r=analyze_review_sentiment(ai,review)results.append(r)# 把结果传回给影刀后续节点result_json=json.dumps(results,ensure_ascii=False)print(result_json)# 影刀会捕获print输出作为节点返回值有什么坑
坑1：API调用频率限制
大模型API都有QPM（每分钟请求数）限制。免费版通常只有10-60次/分钟，如果你在循环里不加控制疯狂请求，很快就会收到429错误。
解决方法：在循环里加time.sleep(1)控制频率。批量处理时用队列 + 逐条消费的模式。
importtimedefbatch_process_with_rate_limit(items,process_func,rpm=30):"""带频率限制的批量处理"""interval=60.0/rpm results=[]foriteminitems:results.append(process_func(item))time.sleep(interval)# 控制请求间隔returnresults坑2：大模型返回结果不稳定
同一个prompt，两次调用可能返回格式略有不同的结果。特别是让模型返回JSON时，它偶尔会在JSON外面包markdown代码块、加多余的空格换行、甚至漏掉引号。
TEMU店群如何管理运营？
解决方法：
- temperature设低一点（0.1-0.3），减少随机性
- 代码里做容错解析，尝试多种方式提取JSON
- 加入重试逻辑，解析失败就重新请求
defrobust_json_parse(ai_response,max_retries=3):"""健壮的JSON解析，带重试"""forattemptinrange(max_retries):try:# 尝试直接解析returnjson.loads(ai_response)exceptjson.JSONDecodeError:# 尝试去掉markdown代码块text=ai_responseif"```"intext:parts=text.split("```")iflen(parts)>=2:text=parts[1]iftext.startswith("json"):text=text[4:]try:returnjson.loads(text.strip())except:ifattempt<max_retries-1:# 重新请求continueelse:return{"error":"JSON解析失败","raw":ai_response[:100]}returnNone坑3：token消耗比你想象的大
每次调用时，system prompt + user message + 返回内容都算token。如果你的system prompt写得像小作文，每条评论只有20个字，但system prompt占了500个token，那就是巨大的浪费。
解决方法：
- system prompt尽量精简，把核心要求写清楚就行
- 长文本输入前先做截断或摘要
- 监控API后台的token消耗，别被自动续费吓到
坑4：网络波动导致超时
大模型API偶尔会因为网络波动导致请求超时。如果你的流程在凌晨跑批量任务，中间一个超时导致整批失败，第二天起来看到报错。
解决方法：设置合理的timeout（建议30-60秒），加上异常捕获和重试机制。不要用默认的超时设置。
坑5：API Key泄露风险
API Key写在Python代码里，如果你把流程导出分享给别人，Key就泄露了。别问我怎么知道的——有人收到过四位数的API账单。
解决方法：把API Key存在影刀的全局变量或配置文件中，代码里用占位符。导出流程前记得清理敏感信息。生产环境建议用环境变量或密钥管理服务。
总结：大模型API + 影刀RPA = 让自动化流程从「机械执行」升级到「智能判断」。关键是控制好调用频率、做好异常处理、保护好API Key。入门推荐先用DeepSeek，便宜好用，踩坑成本低。