dify自动化批量询问LLM并且保存回复为文件在AI应用开发中我们常常需要批量调用大语言模型LLM来处理大量文本数据例如批量生成摘要、批量翻译文档、批量抽取结构化信息等。手动一条条输入不仅效率低下还容易出错。本文将从实战角度出发演示如何使用dify的API接口结合Python脚本实现自动化批量询问LLM并将所有回复保存为文件。### 准备工作dify与API密钥在开始之前请确保你已经部署了dify或使用云端dify并创建了一个应用如“通用对话”应用。从应用设置中获取API密钥API Key和应用IDApp ID。我们将使用dify的chat-messages接口流式/非流式进行批量调用。本示例使用Python 3.8需要安装requests库bashpip install requests### 核心思路循环调用与文件持久化批量询问的本质是准备一个待问列表例如从CSV、JSON或文本文件读取然后循环调用dify API每次传入不同的用户输入最后将每次返回的LLM回复写入一个文件中如.txt或.json。为了便于追踪我们还会记录输入与输出的对应关系。### 代码示例一简单批量询问并保存为文本文件以下脚本从questions.txt读取每行一个问题逐条发送给dify然后将回复追加到answers.txt中格式为“问题 - 回答”。pythonimport requestsimport json# 配置dify API信息DIFY_API_URL https://your-dify-domain.com/api/v1/chat-messages # 替换为你的dify地址API_KEY app-xxxxxxxxxxxxxxxxxxxxxxxxx # 替换为你的API Keydef ask_dify(question: str) - str: 调用dify的非流式chat接口返回LLM的回复文本。 headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { inputs: {}, # 可以传入变量如 {topic: question} query: question, response_mode: blocking, # 非流式等待完整回复 conversation_id: , # 每次新对话 user: batch-user } try: resp requests.post(DIFY_API_URL, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() # 解析回复文本dify blocking模式返回结构中的answer字段 answer data.get(answer, ) return answer except Exception as e: print(f请求失败: {e}) return [ERROR]def main(): # 1. 从文件读取问题列表每行一个问题 questions [] with open(questions.txt, r, encodingutf-8) as f: for line in f: line line.strip() if line: questions.append(line) print(f共读取到 {len(questions)} 个问题开始批量询问...) # 2. 逐条询问并写入结果文件 with open(answers.txt, w, encodingutf-8) as out: for idx, q in enumerate(questions, 1): print(f正在处理第 {idx} 个问题: {q[:30]}...) ans ask_dify(q) # 写入格式问题 - 回答 out.write(f问题 {idx}: {q}\n) out.write(f回答 {idx}: {ans}\n) out.write( * 50 \n) print(所有问题处理完毕结果已保存至 answers.txt)if __name__ __main__: main()使用说明- 在脚本同目录下创建questions.txt每行写一个问题例如什么是人工智能 请用中文解释什么是机器学习。 请写一首关于秋天的诗。- 运行脚本等待完成后查看answers.txt。### 代码示例二批量处理JSON数据并保存为结构化文件实际业务中问题可能附带元数据如ID、类别。以下示例从input.json读取结构化数据每条包含id和question字段调用dify后将回复与原数据组合保存为output.json便于后续分析。pythonimport requestsimport json# 配置dify API信息同上DIFY_API_URL https://your-dify-domain.com/api/v1/chat-messagesAPI_KEY app-xxxxxxxxxxxxxxxxxxxxxxxxxdef ask_dify(question: str) - str: 与示例一相同调用dify并返回回答文本 headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { inputs: {}, query: question, response_mode: blocking, conversation_id: , user: batch-user } try: resp requests.post(DIFY_API_URL, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() return data.get(answer, ) except Exception as e: print(f请求失败: {e}) return [ERROR]def main(): # 1. 读取JSON格式的输入数据 with open(input.json, r, encodingutf-8) as f: records json.load(f) # 预期格式: [{id: 1, question: xxx}, ...] print(f共处理 {len(records)} 条记录) results [] for idx, record in enumerate(records): q record.get(question, ) print(f处理记录 {idx1}: id{record.get(id)}, 问题{q[:30]}...) ans ask_dify(q) # 2. 构建输出记录保留原始字段添加answer output_record { id: record.get(id), question: q, answer: ans, status: success if ans ! [ERROR] else error } results.append(output_record) # 3. 保存为JSON文件 with open(output.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(f处理完成结果已保存至 output.json共 {len(results)} 条)if __name__ __main__: main()input.json示例json[ {id: 101, question: 解释一下量子计算的基本原理。}, {id: 102, question: 写一段关于区块链的科普文字。}, {id: 103, question: 简述TCP/IP协议栈的四层结构。}]运行后output.json将包含每个问题的回答方便后续程序读取或导入数据库。### 进阶优化并发请求与错误重试当问题数量达到几百甚至上千时串行调用会非常慢。我们可以使用concurrent.futures或aiohttp进行并发请求。以下是一个简单的线程池并发版本基于示例一改造pythonfrom concurrent.futures import ThreadPoolExecutor, as_completedimport requests# ... 相同的ask_dify函数 ...def process_question(q: str, idx: int): print(f开始处理第 {idx} 个问题) ans ask_dify(q) return idx, q, ansdef main_concurrent(): questions [] with open(questions.txt, r, encodingutf-8) as f: for line in f: line line.strip() if line: questions.append(line) results [None] * len(questions) # 预分配列表保持顺序 with ThreadPoolExecutor(max_workers5) as executor: # 同时并发5个请求 futures {executor.submit(process_question, q, i): i for i, q in enumerate(questions)} for future in as_completed(futures): idx, q, ans future.result() results[idx] (q, ans) # 按原始索引存放 # 写入文件 with open(answers_concurrent.txt, w, encodingutf-8) as out: for idx, (q, ans) in enumerate(results): out.write(f问题 {idx1}: {q}\n) out.write(f回答 {idx1}: {ans}\n) out.write( * 50 \n) print(并发处理完成结果保存至 answers_concurrent.txt)注意并发数不宜过大以免被dify限流。建议根据dify的部署配置调整max_workers通常5-10比较安全。### 总结本文通过两个实战代码示例展示了如何使用Python调用dify API实现自动化批量询问LLM并将回复保存为文本文件或JSON文件。核心要点包括1.批量循环从文件或JSON读取问题列表逐条调用dify的chat-messages接口。2.结果持久化使用文本文件或JSON文件保存输入与输出便于后续查看或二次处理。3.错误处理通过try/except捕获网络异常避免单个失败中断整个流程。4.并发优化使用线程池提升处理效率适合大规模数据。实际生产环境中还可以加入进度条显示、失败重试机制指数退避、断点续传等功能。希望这篇文章能帮助你快速搭建自己的批量LLM调用工具节省重复劳动的时间。
dify自动化批量询问LLM并且保存回复为文件
dify自动化批量询问LLM并且保存回复为文件在AI应用开发中我们常常需要批量调用大语言模型LLM来处理大量文本数据例如批量生成摘要、批量翻译文档、批量抽取结构化信息等。手动一条条输入不仅效率低下还容易出错。本文将从实战角度出发演示如何使用dify的API接口结合Python脚本实现自动化批量询问LLM并将所有回复保存为文件。### 准备工作dify与API密钥在开始之前请确保你已经部署了dify或使用云端dify并创建了一个应用如“通用对话”应用。从应用设置中获取API密钥API Key和应用IDApp ID。我们将使用dify的chat-messages接口流式/非流式进行批量调用。本示例使用Python 3.8需要安装requests库bashpip install requests### 核心思路循环调用与文件持久化批量询问的本质是准备一个待问列表例如从CSV、JSON或文本文件读取然后循环调用dify API每次传入不同的用户输入最后将每次返回的LLM回复写入一个文件中如.txt或.json。为了便于追踪我们还会记录输入与输出的对应关系。### 代码示例一简单批量询问并保存为文本文件以下脚本从questions.txt读取每行一个问题逐条发送给dify然后将回复追加到answers.txt中格式为“问题 - 回答”。pythonimport requestsimport json# 配置dify API信息DIFY_API_URL https://your-dify-domain.com/api/v1/chat-messages # 替换为你的dify地址API_KEY app-xxxxxxxxxxxxxxxxxxxxxxxxx # 替换为你的API Keydef ask_dify(question: str) - str: 调用dify的非流式chat接口返回LLM的回复文本。 headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { inputs: {}, # 可以传入变量如 {topic: question} query: question, response_mode: blocking, # 非流式等待完整回复 conversation_id: , # 每次新对话 user: batch-user } try: resp requests.post(DIFY_API_URL, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() # 解析回复文本dify blocking模式返回结构中的answer字段 answer data.get(answer, ) return answer except Exception as e: print(f请求失败: {e}) return [ERROR]def main(): # 1. 从文件读取问题列表每行一个问题 questions [] with open(questions.txt, r, encodingutf-8) as f: for line in f: line line.strip() if line: questions.append(line) print(f共读取到 {len(questions)} 个问题开始批量询问...) # 2. 逐条询问并写入结果文件 with open(answers.txt, w, encodingutf-8) as out: for idx, q in enumerate(questions, 1): print(f正在处理第 {idx} 个问题: {q[:30]}...) ans ask_dify(q) # 写入格式问题 - 回答 out.write(f问题 {idx}: {q}\n) out.write(f回答 {idx}: {ans}\n) out.write( * 50 \n) print(所有问题处理完毕结果已保存至 answers.txt)if __name__ __main__: main()使用说明- 在脚本同目录下创建questions.txt每行写一个问题例如什么是人工智能 请用中文解释什么是机器学习。 请写一首关于秋天的诗。- 运行脚本等待完成后查看answers.txt。### 代码示例二批量处理JSON数据并保存为结构化文件实际业务中问题可能附带元数据如ID、类别。以下示例从input.json读取结构化数据每条包含id和question字段调用dify后将回复与原数据组合保存为output.json便于后续分析。pythonimport requestsimport json# 配置dify API信息同上DIFY_API_URL https://your-dify-domain.com/api/v1/chat-messagesAPI_KEY app-xxxxxxxxxxxxxxxxxxxxxxxxxdef ask_dify(question: str) - str: 与示例一相同调用dify并返回回答文本 headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { inputs: {}, query: question, response_mode: blocking, conversation_id: , user: batch-user } try: resp requests.post(DIFY_API_URL, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() return data.get(answer, ) except Exception as e: print(f请求失败: {e}) return [ERROR]def main(): # 1. 读取JSON格式的输入数据 with open(input.json, r, encodingutf-8) as f: records json.load(f) # 预期格式: [{id: 1, question: xxx}, ...] print(f共处理 {len(records)} 条记录) results [] for idx, record in enumerate(records): q record.get(question, ) print(f处理记录 {idx1}: id{record.get(id)}, 问题{q[:30]}...) ans ask_dify(q) # 2. 构建输出记录保留原始字段添加answer output_record { id: record.get(id), question: q, answer: ans, status: success if ans ! [ERROR] else error } results.append(output_record) # 3. 保存为JSON文件 with open(output.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) print(f处理完成结果已保存至 output.json共 {len(results)} 条)if __name__ __main__: main()input.json示例json[ {id: 101, question: 解释一下量子计算的基本原理。}, {id: 102, question: 写一段关于区块链的科普文字。}, {id: 103, question: 简述TCP/IP协议栈的四层结构。}]运行后output.json将包含每个问题的回答方便后续程序读取或导入数据库。### 进阶优化并发请求与错误重试当问题数量达到几百甚至上千时串行调用会非常慢。我们可以使用concurrent.futures或aiohttp进行并发请求。以下是一个简单的线程池并发版本基于示例一改造pythonfrom concurrent.futures import ThreadPoolExecutor, as_completedimport requests# ... 相同的ask_dify函数 ...def process_question(q: str, idx: int): print(f开始处理第 {idx} 个问题) ans ask_dify(q) return idx, q, ansdef main_concurrent(): questions [] with open(questions.txt, r, encodingutf-8) as f: for line in f: line line.strip() if line: questions.append(line) results [None] * len(questions) # 预分配列表保持顺序 with ThreadPoolExecutor(max_workers5) as executor: # 同时并发5个请求 futures {executor.submit(process_question, q, i): i for i, q in enumerate(questions)} for future in as_completed(futures): idx, q, ans future.result() results[idx] (q, ans) # 按原始索引存放 # 写入文件 with open(answers_concurrent.txt, w, encodingutf-8) as out: for idx, (q, ans) in enumerate(results): out.write(f问题 {idx1}: {q}\n) out.write(f回答 {idx1}: {ans}\n) out.write( * 50 \n) print(并发处理完成结果保存至 answers_concurrent.txt)注意并发数不宜过大以免被dify限流。建议根据dify的部署配置调整max_workers通常5-10比较安全。### 总结本文通过两个实战代码示例展示了如何使用Python调用dify API实现自动化批量询问LLM并将回复保存为文本文件或JSON文件。核心要点包括1.批量循环从文件或JSON读取问题列表逐条调用dify的chat-messages接口。2.结果持久化使用文本文件或JSON文件保存输入与输出便于后续查看或二次处理。3.错误处理通过try/except捕获网络异常避免单个失败中断整个流程。4.并发优化使用线程池提升处理效率适合大规模数据。实际生产环境中还可以加入进度条显示、失败重试机制指数退避、断点续传等功能。希望这篇文章能帮助你快速搭建自己的批量LLM调用工具节省重复劳动的时间。
