Flowise开发者实践Python脚本调用Flowise API1. 开篇为什么需要Python调用Flowise如果你已经用Flowise搭建好了AI工作流比如一个智能客服机器人或者文档问答系统接下来最实际的问题就是怎么让我的Python程序调用这个工作流想象一下这样的场景你的网站用户提问你需要把问题发送给Flowise工作流拿到智能回复后再展示给用户。这个发送问题-获取回复的过程就需要通过API调用来实现。本文将手把手教你如何用Python脚本调用Flowise API让你搭建的AI工作流真正融入到实际应用中。2. 准备工作获取API调用信息在开始写代码之前我们需要先准备好三样东西2.1 获取Flowise API端点首先打开你的Flowise界面找到已经搭建好的工作流。在画布右上角点击设置图标你会看到API相关信息API端点通常格式是http://你的服务器地址:3000/api/v1/prediction/工作流IDAPI密钥如果需要认证的话可以在环境变量中设置2.2 安装必要的Python库你需要的主要是requests库用于发送HTTP请求pip install requests如果还需要处理JSON数据Python自带的json库就足够了。2.3 了解API参数不同的工作流需要的输入参数可能不同。通常你需要提供用户的问题或输入文本可选的对话历史其他特定参数如文件、配置选项等3. 基础调用最简单的API请求让我们从最简单的例子开始看看如何用Python发送请求到Flowise工作流。3.1 基本请求代码import requests import json # Flowise API配置 FLOWISE_URL http://localhost:3000/api/v1/prediction/你的工作流ID API_KEY 你的API密钥 # 如果设置了认证 def call_flowise_simple(question): 调用Flowise工作流的简单示例 headers { Content-Type: application/json } # 如果有API密钥添加到headers if API_KEY: headers[Authorization] fBearer {API_KEY} # 请求数据 payload { question: question } try: response requests.post(FLOWISE_URL, headersheaders, jsonpayload) response.raise_for_status() # 检查请求是否成功 result response.json() return result except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None # 测试调用 if __name__ __main__: result call_flowise_simple(你好请介绍一下你自己) if result: print(Flowise回复:, result)3.2 处理响应数据Flowise的响应通常是JSON格式包含工作流的输出# 假设响应格式如下 { text: 这是AI的回复内容, sourceDocuments: [...], # 如果是RAG工作流可能包含参考文档 otherData: ... # 其他自定义输出 } def process_response(response): 处理Flowise的响应 if not response: return 抱歉服务暂时不可用 # 提取主要回复文本 reply_text response.get(text, 未获取到回复) # 如果有参考文档可以一并处理 sources response.get(sourceDocuments, []) if sources: reply_text \n\n参考资料 for source in sources[:3]: # 只显示前3个参考 reply_text f\n- {source.get(metadata, {}).get(source, 未知来源)} return reply_text4. 实战进阶处理复杂场景在实际应用中你可能需要处理更复杂的场景。下面看几个常见情况的处理方法。4.1 带对话历史的连续对话很多聊天机器人需要维护对话上下文def call_flowise_with_history(question, chat_historyNone): 带对话历史的API调用 headers { Content-Type: application/json } if API_KEY: headers[Authorization] fBearer {API_KEY} payload { question: question, history: chat_history or [] # 传入对话历史 } try: response requests.post(FLOWISE_URL, headersheaders, jsonpayload) response.raise_for_status() result response.json() # 更新对话历史 new_history (chat_history or []) [ {role: user, content: question}, {role: assistant, content: result.get(text, )} ] return result, new_history except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None, chat_history # 使用示例 chat_history [] user_question 你好 response, chat_history call_flowise_with_history(user_question, chat_history) print(AI回复:, response.get(text)) # 继续对话 next_question 刚才我们说了什么 response, chat_history call_flowise_with_history(next_question, chat_history) print(AI回复:, response.get(text))4.2 处理文件上传和RAG场景如果你的Flowise工作流涉及文档处理def call_flowise_with_file(question, file_path): 调用支持文件处理的工作流 headers { Authorization: fBearer {API_KEY} if API_KEY else } # 构建multipart/form-data请求 files { file: (open(file_path, rb)), question: (None, question) } try: response requests.post(FLOWISE_URL, headersheaders, filesfiles) response.raise_for_status() return response.json() except Exception as e: print(f文件处理失败: {e}) return None4.3 异步调用提高性能对于需要高并发的场景可以使用异步请求import aiohttp import asyncio async def async_call_flowise(session, question): 异步调用Flowise API payload {question: question} try: async with session.post(FLOWISE_URL, jsonpayload) as response: response.raise_for_status() return await response.json() except Exception as e: print(f异步请求失败: {e}) return None async def batch_call_questions(questions): 批量处理多个问题 async with aiohttp.ClientSession() as session: tasks [async_call_flowise(session, q) for q in questions] results await asyncio.gather(*tasks) return results # 使用示例 questions [问题1, 问题2, 问题3] results asyncio.run(batch_call_questions(questions))5. 错误处理与最佳实践在实际生产环境中健壮的错误处理至关重要。5.1 完善的错误处理def robust_flowise_call(question, max_retries3): 带重试机制的API调用 headers { Content-Type: application/json } if API_KEY: headers[Authorization] fBearer {API_KEY} payload {question: question} for attempt in range(max_retries): try: response requests.post( FLOWISE_URL, headersheaders, jsonpayload, timeout30 # 设置超时时间 ) # 检查HTTP状态码 if response.status_code 200: return response.json() elif response.status_code 429: # 速率限制等待后重试 wait_time 2 ** attempt # 指数退避 print(f速率限制等待{wait_time}秒后重试...) time.sleep(wait_time) continue else: response.raise_for_status() except requests.exceptions.Timeout: print(f请求超时尝试 {attempt 1}/{max_retries}) except requests.exceptions.RequestException as e: print(f请求异常: {e}) break return {text: 服务暂时不可用请稍后再试} # 添加超时控制 import signal class TimeoutException(Exception): pass def timeout_handler(signum, frame): raise TimeoutException(API调用超时) def call_with_timeout(question, timeout30): 带超时控制的API调用 signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout) try: result robust_flowise_call(question) signal.alarm(0) # 取消闹钟 return result except TimeoutException: return {text: 请求超时请检查网络连接或稍后再试}5.2 性能监控和日志记录import time import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def monitored_flowise_call(question): 带监控的API调用 start_time time.time() try: result robust_flowise_call(question) end_time time.time() # 记录性能指标 latency end_time - start_time logger.info(fAPI调用成功 - 耗时: {latency:.2f}秒) # 可以在这里添加监控数据上报 # track_metrics(latency, True) return result except Exception as e: end_time time.time() latency end_time - start_time logger.error(fAPI调用失败 - 耗时: {latency:.2f}秒 - 错误: {e}) # track_metrics(latency, False) return None6. 实际应用案例让我们看几个实际的应用场景了解如何将Flowise API集成到真实项目中。6.1 集成到Web应用from flask import Flask, request, jsonify app Flask(__name__) app.route(/chat, methods[POST]) def chat_endpoint(): Web应用的聊天接口 data request.json user_message data.get(message, ) session_id data.get(session_id, default) if not user_message: return jsonify({error: 消息不能为空}), 400 # 这里可以从数据库或缓存中获取对话历史 # chat_history get_chat_history(session_id) # 调用Flowise API response robust_flowise_call(user_message) if response: # 保存对话历史到数据库 # save_to_history(session_id, user_message, response.get(text, )) return jsonify({ reply: response.get(text, 未获取到回复), sources: response.get(sourceDocuments, []) }) else: return jsonify({error: 服务暂时不可用}), 503 if __name__ __main__: app.run(debugTrue)6.2 批量处理任务import pandas as pd from tqdm import tqdm def batch_process_csv(input_file, output_file): 批量处理CSV文件中的问题 # 读取数据 df pd.read_csv(input_file) results [] for index, row in tqdm(df.iterrows(), totallen(df)): question row[question] # 调用Flowise API response robust_flowise_call(question) if response: results.append({ question: question, answer: response.get(text, ), sources: str(response.get(sourceDocuments, [])) }) else: results.append({ question: question, answer: 处理失败, sources: }) # 避免请求过于频繁 time.sleep(0.5) # 保存结果 result_df pd.DataFrame(results) result_df.to_csv(output_file, indexFalse) print(f处理完成结果已保存到 {output_file})7. 总结通过本文的学习你应该已经掌握了如何使用Python脚本调用Flowise API的各种技巧。让我们简单回顾一下重点核心要点使用requests库发送POST请求到Flowise的API端点正确处理认证、参数传递和响应解析实现对话历史维护以支持连续对话添加完善的错误处理和重试机制最佳实践总是添加超时控制和错误处理对于生产环境实现监控和日志记录考虑使用异步请求提高并发性能遵守API速率限制避免过度请求下一步建议尝试将Flowise API集成到你自己的项目中探索更复杂的工作流和参数配置考虑添加缓存机制减少重复请求实现负载均衡和多实例部署以提高可靠性现在你已经具备了将Flowise工作流集成到Python应用中的能力可以开始构建更强大的AI应用了获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。