影刀RPA 调用大模型API实战让自动化流程拥有AI大脑作者林焱什么情况用在RPA自动化的过程中你经常会遇到一些「规则搞不定」的情况抓取了一批商品评论需要判断每条是好评还是差评爬到了几十篇文章标题想自动归纳出核心观点收到客户邮件需要判断紧急程度并分类转发批量生成商品描述、邮件回复、会议纪要等文本内容传统RPA做这些事情基本没戏——它不是靠规则能解决的。但把这些任务交给大模型GPT/Claude/文心一言/通义千问配合影刀的Python节点调用API你的自动化流程就像突然长出了大脑。核心场景需要判断、理解、归纳、生成文本内容的自动化任务。怎么做第一步搞到大模型的API Key不管你用的是哪家大模型都需要先拿到API Key。以国内最常用的几个为例通义千问阿里云打开阿里云百炼平台bailian.console.aliyun.com创建一个应用获取API Key模型名用qwen-turbo便宜或qwen-plus效果好拼多多店群自动化上架方案DeepSeek5. 访问 platform.deepseek.com6. 注册后创建API Key7. 价格极低效果很好推荐新手用费用方面不用太担心——调用一次通常几分钱处理一千条评论也就几块钱。第二步在影刀里封装调用函数影刀的Python节点支持直接发HTTP请求我们封装一个通用的大模型调用函数。importrequestsimportjsonclassAICaller: 通用大模型调用类 支持OpenAI兼容的API格式通义千问/DeepSeek/GLM等都兼容 def__init__(self,api_key,base_url,modelqwen-turbo):self.api_keyapi_key self.base_urlbase_url.rstrip(/)self.modelmodeldefchat(self,system_prompt,user_message,temperature0.7): 单轮对话发一个system prompt和user message拿回结果 urlf{self.base_url}/chat/completionsheaders{Authorization:fBearer{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:resprequests.post(url,headersheaders,jsonpayload,timeout60)resp.raise_for_status()resultresp.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_msgf请将以下文本分类到这些类别之一{cat_list}返回格式只返回JSON不要有其他内容 {{category: 分类结果, confidence: 高/中/低}} 待分类文本{text}resultself.chat(system_prompt,user_msg,temperature0.1)returnresult# 初始化填入你自己的API Key aiAICaller(api_keysk-your-api-key-here,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,modelqwen-turbo)第三步实战——商品评论情感分析这是最常见的需求。拿到一堆评论自动标注好差评。defanalyze_review_sentiment(ai,review_text):分析单条评论的情感system_prompt你是一个电商评论分析专家。 请分析用户评论的情感倾向返回JSON格式 {sentiment: 正面/负面/中性, reason: 简短理由} 注意如果评论包含明显的不满、质量问题、物流投诉判定为负面。resultai.chat(system_prompt,review_text,temperature0.1)# 解析JSON结果处理可能的格式异常try:# 大模型有时候会在JSON外面包markdown代码块ifinresult:resultresult.split()[1]ifresult.startswith(json):resultresult[4:]resultresult.strip()returnjson.loads(result)exceptjson.JSONDecodeError:return{sentiment:无法判断,reason:f解析失败:{result[:50]}}# 示例调用reviews[物流太慢了等了5天才到包装也破了,质量很好比想象中还要好下次还来,还行吧无功无过,客服态度极差问题拖了三天没解决]forreviewinreviews:resultanalyze_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_msgf邮件标题{email_subject}\n\n邮件正文\n{email_body}resultai.chat(system_prompt,user_msg,temperature0.1)try:ifinresult:resultresult.split()[1]ifresult.startswith(json):resultresult[4:]resultresult.strip()returnjson.loads(result)except:return{department:其他,urgency:普通,summary:解析失败}# 示例email_resultclassify_email(ai,紧急服务器502错误客户无法下单,刚才收到客户投诉官网下单页面一直报502错误已经持续半小时了请尽快处理)print(json.dumps(email_result,ensure_asciiFalse,indent2))第五步实战——批量生成SEO标题和描述电商运营每天要写几百个商品标题让大模型批量生成。defgenerate_seo_titles(ai,product_info,count5):根据产品信息生成多个SEO友好的标题system_prompt你是电商SEO专家。请根据产品信息生成吸引人的商品标题。 要求 1. 包含核心关键词 2. 长度15-30字 3. 突出卖点和优惠 4. 生成指定数量用换行分隔user_msgf产品信息{product_info}请生成{count}个SEO友好的商品标题。resultai.chat(system_prompt,user_msg,temperature0.8)titles[t.strip(- ).strip()fortinresult.split(\n)ift.strip()]returntitles[:count]# 示例titlesgenerate_seo_titles(ai,女士连衣裙雪纺面料夏款收腰设计限时优惠价199元)fori,tinenumerate(titles,1):print(f{i}.{t})第六步在影刀流程中串联把上面的Python代码放到影刀的Python节点里输入来自前面的网页抓取节点输出传给Excel写入节点。影刀流程图 ┌──────────────┐ │ 网页抓取节点 │ → 获取商品评论列表 └──────┬───────┘ ↓ ┌──────────────┐ │ Python节点① │ → 循环调用AI判断情感 └──────┬───────┘ ↓ ┌──────────────┐ │ 条件分支节点 │ → 负面评论 → 企微告警 └──────┬───────┘ 正面评论 → 写入Excel ↓ ┌──────────────┐ │ Excel写入节点 │ → 汇总到情感分析报告 └──────────────┘Python节点里接收影刀变量的写法# 影刀传过来的变量通过全局变量获取# 假设影刀里设置了变量 review_listimportjson reviewsjson.loads(review_list)# 影刀传来的JSON字符串results[]forreviewinreviews:ranalyze_review_sentiment(ai,review)results.append(r)# 把结果传回给影刀后续节点result_jsonjson.dumps(results,ensure_asciiFalse)print(result_json)# 影刀会捕获print输出作为节点返回值有什么坑坑1API调用频率限制大模型API都有QPM每分钟请求数限制。免费版通常只有10-60次/分钟如果你在循环里不加控制疯狂请求很快就会收到429错误。解决方法在循环里加time.sleep(1)控制频率。批量处理时用队列 逐条消费的模式。importtimedefbatch_process_with_rate_limit(items,process_func,rpm30):带频率限制的批量处理interval60.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_retries3):健壮的JSON解析带重试forattemptinrange(max_retries):try:# 尝试直接解析returnjson.loads(ai_response)exceptjson.JSONDecodeError:# 尝试去掉markdown代码块textai_responseifintext:partstext.split()iflen(parts)2:textparts[1]iftext.startswith(json):texttext[4:]try:returnjson.loads(text.strip())except:ifattemptmax_retries-1:# 重新请求continueelse:return{error:JSON解析失败,raw:ai_response[:100]}returnNone坑3token消耗比你想象的大每次调用时system prompt user message 返回内容都算token。如果你的system prompt写得像小作文每条评论只有20个字但system prompt占了500个token那就是巨大的浪费。解决方法system prompt尽量精简把核心要求写清楚就行长文本输入前先做截断或摘要监控API后台的token消耗别被自动续费吓到坑4网络波动导致超时大模型API偶尔会因为网络波动导致请求超时。如果你的流程在凌晨跑批量任务中间一个超时导致整批失败第二天起来看到报错。解决方法设置合理的timeout建议30-60秒加上异常捕获和重试机制。不要用默认的超时设置。坑5API Key泄露风险API Key写在Python代码里如果你把流程导出分享给别人Key就泄露了。别问我怎么知道的——有人收到过四位数的API账单。解决方法把API Key存在影刀的全局变量或配置文件中代码里用占位符。导出流程前记得清理敏感信息。生产环境建议用环境变量或密钥管理服务。总结大模型API 影刀RPA 让自动化流程从「机械执行」升级到「智能判断」。关键是控制好调用频率、做好异常处理、保护好API Key。入门推荐先用DeepSeek便宜好用踩坑成本低。
影刀RPA 调用大模型API实战：让自动化流程拥有AI大脑
影刀RPA 调用大模型API实战让自动化流程拥有AI大脑作者林焱什么情况用在RPA自动化的过程中你经常会遇到一些「规则搞不定」的情况抓取了一批商品评论需要判断每条是好评还是差评爬到了几十篇文章标题想自动归纳出核心观点收到客户邮件需要判断紧急程度并分类转发批量生成商品描述、邮件回复、会议纪要等文本内容传统RPA做这些事情基本没戏——它不是靠规则能解决的。但把这些任务交给大模型GPT/Claude/文心一言/通义千问配合影刀的Python节点调用API你的自动化流程就像突然长出了大脑。核心场景需要判断、理解、归纳、生成文本内容的自动化任务。怎么做第一步搞到大模型的API Key不管你用的是哪家大模型都需要先拿到API Key。以国内最常用的几个为例通义千问阿里云打开阿里云百炼平台bailian.console.aliyun.com创建一个应用获取API Key模型名用qwen-turbo便宜或qwen-plus效果好拼多多店群自动化上架方案DeepSeek5. 访问 platform.deepseek.com6. 注册后创建API Key7. 价格极低效果很好推荐新手用费用方面不用太担心——调用一次通常几分钱处理一千条评论也就几块钱。第二步在影刀里封装调用函数影刀的Python节点支持直接发HTTP请求我们封装一个通用的大模型调用函数。importrequestsimportjsonclassAICaller: 通用大模型调用类 支持OpenAI兼容的API格式通义千问/DeepSeek/GLM等都兼容 def__init__(self,api_key,base_url,modelqwen-turbo):self.api_keyapi_key self.base_urlbase_url.rstrip(/)self.modelmodeldefchat(self,system_prompt,user_message,temperature0.7): 单轮对话发一个system prompt和user message拿回结果 urlf{self.base_url}/chat/completionsheaders{Authorization:fBearer{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:resprequests.post(url,headersheaders,jsonpayload,timeout60)resp.raise_for_status()resultresp.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_msgf请将以下文本分类到这些类别之一{cat_list}返回格式只返回JSON不要有其他内容 {{category: 分类结果, confidence: 高/中/低}} 待分类文本{text}resultself.chat(system_prompt,user_msg,temperature0.1)returnresult# 初始化填入你自己的API Key aiAICaller(api_keysk-your-api-key-here,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,modelqwen-turbo)第三步实战——商品评论情感分析这是最常见的需求。拿到一堆评论自动标注好差评。defanalyze_review_sentiment(ai,review_text):分析单条评论的情感system_prompt你是一个电商评论分析专家。 请分析用户评论的情感倾向返回JSON格式 {sentiment: 正面/负面/中性, reason: 简短理由} 注意如果评论包含明显的不满、质量问题、物流投诉判定为负面。resultai.chat(system_prompt,review_text,temperature0.1)# 解析JSON结果处理可能的格式异常try:# 大模型有时候会在JSON外面包markdown代码块ifinresult:resultresult.split()[1]ifresult.startswith(json):resultresult[4:]resultresult.strip()returnjson.loads(result)exceptjson.JSONDecodeError:return{sentiment:无法判断,reason:f解析失败:{result[:50]}}# 示例调用reviews[物流太慢了等了5天才到包装也破了,质量很好比想象中还要好下次还来,还行吧无功无过,客服态度极差问题拖了三天没解决]forreviewinreviews:resultanalyze_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_msgf邮件标题{email_subject}\n\n邮件正文\n{email_body}resultai.chat(system_prompt,user_msg,temperature0.1)try:ifinresult:resultresult.split()[1]ifresult.startswith(json):resultresult[4:]resultresult.strip()returnjson.loads(result)except:return{department:其他,urgency:普通,summary:解析失败}# 示例email_resultclassify_email(ai,紧急服务器502错误客户无法下单,刚才收到客户投诉官网下单页面一直报502错误已经持续半小时了请尽快处理)print(json.dumps(email_result,ensure_asciiFalse,indent2))第五步实战——批量生成SEO标题和描述电商运营每天要写几百个商品标题让大模型批量生成。defgenerate_seo_titles(ai,product_info,count5):根据产品信息生成多个SEO友好的标题system_prompt你是电商SEO专家。请根据产品信息生成吸引人的商品标题。 要求 1. 包含核心关键词 2. 长度15-30字 3. 突出卖点和优惠 4. 生成指定数量用换行分隔user_msgf产品信息{product_info}请生成{count}个SEO友好的商品标题。resultai.chat(system_prompt,user_msg,temperature0.8)titles[t.strip(- ).strip()fortinresult.split(\n)ift.strip()]returntitles[:count]# 示例titlesgenerate_seo_titles(ai,女士连衣裙雪纺面料夏款收腰设计限时优惠价199元)fori,tinenumerate(titles,1):print(f{i}.{t})第六步在影刀流程中串联把上面的Python代码放到影刀的Python节点里输入来自前面的网页抓取节点输出传给Excel写入节点。影刀流程图 ┌──────────────┐ │ 网页抓取节点 │ → 获取商品评论列表 └──────┬───────┘ ↓ ┌──────────────┐ │ Python节点① │ → 循环调用AI判断情感 └──────┬───────┘ ↓ ┌──────────────┐ │ 条件分支节点 │ → 负面评论 → 企微告警 └──────┬───────┘ 正面评论 → 写入Excel ↓ ┌──────────────┐ │ Excel写入节点 │ → 汇总到情感分析报告 └──────────────┘Python节点里接收影刀变量的写法# 影刀传过来的变量通过全局变量获取# 假设影刀里设置了变量 review_listimportjson reviewsjson.loads(review_list)# 影刀传来的JSON字符串results[]forreviewinreviews:ranalyze_review_sentiment(ai,review)results.append(r)# 把结果传回给影刀后续节点result_jsonjson.dumps(results,ensure_asciiFalse)print(result_json)# 影刀会捕获print输出作为节点返回值有什么坑坑1API调用频率限制大模型API都有QPM每分钟请求数限制。免费版通常只有10-60次/分钟如果你在循环里不加控制疯狂请求很快就会收到429错误。解决方法在循环里加time.sleep(1)控制频率。批量处理时用队列 逐条消费的模式。importtimedefbatch_process_with_rate_limit(items,process_func,rpm30):带频率限制的批量处理interval60.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_retries3):健壮的JSON解析带重试forattemptinrange(max_retries):try:# 尝试直接解析returnjson.loads(ai_response)exceptjson.JSONDecodeError:# 尝试去掉markdown代码块textai_responseifintext:partstext.split()iflen(parts)2:textparts[1]iftext.startswith(json):texttext[4:]try:returnjson.loads(text.strip())except:ifattemptmax_retries-1:# 重新请求continueelse:return{error:JSON解析失败,raw:ai_response[:100]}returnNone坑3token消耗比你想象的大每次调用时system prompt user message 返回内容都算token。如果你的system prompt写得像小作文每条评论只有20个字但system prompt占了500个token那就是巨大的浪费。解决方法system prompt尽量精简把核心要求写清楚就行长文本输入前先做截断或摘要监控API后台的token消耗别被自动续费吓到坑4网络波动导致超时大模型API偶尔会因为网络波动导致请求超时。如果你的流程在凌晨跑批量任务中间一个超时导致整批失败第二天起来看到报错。解决方法设置合理的timeout建议30-60秒加上异常捕获和重试机制。不要用默认的超时设置。坑5API Key泄露风险API Key写在Python代码里如果你把流程导出分享给别人Key就泄露了。别问我怎么知道的——有人收到过四位数的API账单。解决方法把API Key存在影刀的全局变量或配置文件中代码里用占位符。导出流程前记得清理敏感信息。生产环境建议用环境变量或密钥管理服务。总结大模型API 影刀RPA 让自动化流程从「机械执行」升级到「智能判断」。关键是控制好调用频率、做好异常处理、保护好API Key。入门推荐先用DeepSeek便宜好用踩坑成本低。
