为了实现一个由“用户提问”驱动，自动化完成“AI联网搜索”、“爬取网上RSS信息”，并最终使用“智谱AI整合生成文章”的工作流，我们可以设计一个分步执行的Python脚本或简易的自动化流程。本方案将整合联网搜索、RSS解析、文章摘要及智能写作功能，力求步骤清晰、代码简洁，适合初学者学习实现。 ### **一、 工作流整体设计** 整个工作流可以划分为四个核心步骤，构成一个线性的自动化管道 [ref_3]： | 步骤 | 核心功能 | 关键技术与API | 输出物 | | :--- | :--- | :--- | :--- | | **1. 用户输入与关键词生成** | 接收用户原始问题，解析并提炼搜索引擎友好的关键词。 | 大型语言模型 (LLM) | 优化的搜索关键词 | | **2. 联网搜索与信息获取** | 基于关键词执行联网搜索，获取相关网页信息。 | Tavily Search API | 包含摘要和链接的搜索结果列表 | | **3. RSS信息抓取与处理** | 根据搜索结果或预设的RSS源，抓取结构化信息。 | `feedparser` 库 | 从RSS源获取的文章条目列表 | | **4. 信息整合与文章生成** | 将搜索和RSS信息作为素材，调用大模型撰写、润色成文。 | 智谱AI GLM API | 一篇完整的、带有引用的文章 | ### **二、 环境准备与关键库安装** 在开始编写代码前，需要准备必要的API密钥并安装Python库。 1. **申请API密钥**: * **Tavily API Key**: 用于联网搜索。前往 [Tavily官网](https://app.tavily.com) 注册并获取。 * **智谱AI API Key**: 用于文章生成。前往 [智谱AI开放平台](https://open.bigmodel.cn/) 注册并获取 [ref_4][ref_5]。 2. **安装Python库**: 打开终端或命令提示符，执行以下命令安装所需依赖： ```bash pip install requests feedparser langchain-community ``` * `requests`: 用于发起HTTP请求，调用各种API [ref_6]。 * `feedparser`: 一个非常简洁易用的库，用于解析RSS/Atom源。 * `langchain-community`: 这里主要使用其封装的 `TavilySearchResults` 工具，简化搜索调用 [ref_2]。你也可以直接使用Tavily的原始REST API。 ### **三、 分步代码实现** 以下代码将上述四个步骤串联起来。请将代码中的 `YOUR_TAVILY_API_KEY` 和 `YOUR_ZHIPUAI_API_KEY` 替换为你自己的密钥。 ```python # -*- coding: utf-8 -*- """ AI联网搜索 + RSS爬取 + 智谱AI整合写作工作流 目标：用户输入问题 -> 联网搜索 -> 抓取RSS -> 生成文章 """ import os import json import requests import feedparser from typing import List, Dict from langchain_community.tools.tavily_search import TavilySearchResults # ==================== 第一步：配置环境变量 (请在此处填入你的API密钥) ==================== TAVILY_API_KEY = "YOUR_TAVILY_API_KEY" # 替换为你的Tavily API Key ZHIPUAI_API_KEY = "YOUR_ZHIPUAI_API_KEY" # 替换为你的智谱AI API Key # 设置智谱AI API的基础URL ZHIPUAI_API_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions" # ==================== 第二步：联网搜索功能 ==================== def perform_web_search(query: str, max_results: int = 5) -> List[Dict]: """ 使用Tavily进行联网搜索。 :param query: 搜索关键词/问题 :param max_results: 返回的最大结果数量 :return: 包含搜索结果的字典列表，每个字典包含 `title`, `url`, `content` (摘要) 等字段 [ref_2]。 """ print(f"正在联网搜索: {query}") try: # 初始化Tavily搜索工具 search_tool = TavilySearchResults(tavily_api_key=TAVILY_API_KEY, max_results=max_results) # 执行搜索 search_results = search_tool.invoke(query) print(f"搜索完成，找到 {len(search_results)} 条结果。") return search_results except Exception as e: print(f"搜索过程出错: {e}") return [] # ==================== 第三步：RSS信息抓取功能 ==================== def fetch_rss_feed(rss_url: str, max_entries: int = 10) -> List[Dict]: """ 解析指定的RSS源，获取最新文章信息。 :param rss_url: RSS源的URL地址 :param max_entries: 获取的最大条目数 :return: 包含RSS文章信息的字典列表，每个字典包含 `title`, `link`, `summary` 等字段。 """ print(f"正在抓取RSS源: {rss_url}") feed_info = [] try: # 使用feedparser解析RSS feed = feedparser.parse(rss_url) for entry in feed.entries[:max_entries]: item = { "title": entry.get("title", "无标题"), "link": entry.get("link", "#"), # RSS中摘要可能在 `summary` 或 `description` 字段 "summary": entry.get("summary") or entry.get("description", "无摘要"), "published": entry.get("published", "未知日期") } feed_info.append(item) print(f"RSS抓取完成，获取到 {len(feed_info)} 篇文章。") except Exception as e: print(f"RSS抓取出错: {e}") return feed_info # ==================== 第四步：调用智谱AI生成文章 ==================== def generate_article_with_zhipuai(user_query: str, search_context: List[Dict], rss_context: List[Dict]) -> str: """ 整合搜索和RSS信息，调用智谱AI GLM模型生成结构化的文章 [ref_4][ref_5]。 :param user_query: 用户的原始问题 :param search_context: 联网搜索结果列表 :param rss_context: RSS抓取结果列表 :return: 智谱AI生成的文章内容 """ print("正在调用智谱AI整合信息并生成文章...") # 1. 构建模型的系统提示 (System Prompt)，指导AI扮演的角色和任务 system_prompt = """你是一位专业的AI助手。你的任务是根据用户提供的原始问题、联网搜索的结果摘要以及相关RSS信息源，撰写一篇结构清晰、信息准确、语言流畅的文章。 文章要求： 1. 开头直接回应用户的问题。 2. 正文部分有机地融合搜索结果和RSS信息中的事实和观点，并注明信息出处（例如，提及“根据XX网站报道”或“搜索结果显示”）。 3. 结尾进行总结或提出展望。 请直接输出文章正文，无需在开头说“好的”或进行其他寒暄。""" # 2. 将搜索和RSS信息格式化为易于模型理解的文本 formatted_search_info = "\n".join([f"- [{res.get('title', 'N/A')}]({res.get('url', '#')}): {res.get('content', '无摘要')}" for res in search_context]) formatted_rss_info = "\n".join([f"- [{entry.get('title', 'N/A')}]({entry.get('link', '#')}): {entry.get('summary', '无摘要')} (发布时间: {entry.get('published')})" for entry in rss_context]) # 3. 构建用户消息，包含所有上下文 user_message = f""" **用户原始问题**：{user_query} **联网搜索获得的关键信息（共{len(search_context)}条）**： {formatted_search_info if formatted_search_info else '（本次搜索未获得有效结果）'} **从相关RSS源获取的最新动态（共{len(rss_context)}条）**： {formatted_rss_info if formatted_rss_info else '（本次未提供RSS源或RSS源无效）'} 请基于以上所有信息，撰写文章。 """ # 4. 准备请求智谱AI API的载荷 (Payload) headers = { "Authorization": f"Bearer {ZHIPUAI_API_KEY}", "Content-Type": "application/json" } payload = { "model": "glm-4", # 使用GLM-4模型，可根据需要更换为 `glm-4-air` 或 `glm-4-flash` 等 [ref_5] "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ], # 以下参数可根据需要调整 "temperature": 0.7, # 控制创造性，越低越稳定，越高越随机 "max_tokens": 2000 # 控制生成文章的最大长度 } # 5. 发起API请求并获取响应 try: response = requests.post(ZHIPUAI_API_URL, headers=headers, json=payload, timeout=60) response.raise_for_status() # 检查HTTP请求是否成功 result = response.json() # 解析智谱AI的响应，提取生成的内容 ai_generated_content = result["choices"][0]["message"]["content"] print("文章生成成功！") return ai_generated_content except requests.exceptions.RequestException as e: print(f"调用智谱AI API时发生网络错误: {e}") return f"文章生成失败，API调用错误：{e}" except (KeyError, IndexError) as e: print(f"解析智谱AI API响应时出错: {e}, 原始响应: {result}") return "文章生成失败，响应格式异常。" # ==================== 第五步：主函数，串联整个工作流 ==================== def main_workflow(): """ 主工作流：依次执行用户输入、搜索、RSS抓取、文章生成。 """ print("=" * 50) print("AI联网搜索+RSS整合写作工作流") print("=" * 50) # 1. 用户输入问题 user_question = input("请输入您想了解的问题（例如：'人工智能在医疗领域的最新进展'）: ").strip() if not user_question: print("问题不能为空。") return # 2. 执行联网搜索 (关键词直接用用户问题，也可在此前增加一个LLM提炼关键词的步骤 [ref_3]) search_results = perform_web_search(user_question) # 3. 抓取RSS信息 (这里以“机器之心”的RSS为例，你可以替换成任何与问题相关的RSS源) # 示例RSS源：机器之心 - 人工智能领域资讯 relevant_rss_url = "https://www.jiqizhixin.com/rss" # 你也可以根据搜索结果的域名，动态选择或添加RSS源，此处为简化流程使用固定源。 rss_articles = fetch_rss_feed(relevant_rss_url) # 4. 调用智谱AI，整合信息并生成文章 final_article = generate_article_with_zhipuai(user_question, search_results, rss_articles) # 5. 输出最终结果 print("\n" + "=" * 50) print("【生成的文章】") print("=" * 50) print(final_article) print("=" * 50) # （可选）将文章保存到文件 save_to_file = input("\n是否将文章保存到文件？(输入 y 保存，其他键跳过): ").lower() if save_to_file == 'y': filename = f"generated_article_{user_question[:20]}.txt".replace(' ', '_').replace('/', '_') with open(filename, 'w', encoding='utf-8') as f: f.write(final_article) print(f"文章已保存至: {filename}") # ==================== 程序入口 ==================== if __name__ == "__main__": main_workflow() ``` ### **四、 运行与使用说明** 1. **保存代码**：将上面的代码完整复制，保存为一个 `.py` 文件，例如 `ai_writer_workflow.py`。 2. **配置密钥**：务必修改代码开头的 `TAVILY_API_KEY` 和 `ZHIPUAI_API_KEY` 为你自己的密钥。 3. **运行脚本**：在终端中，导航到脚本所在目录，运行命令： ```bash python ai_writer_workflow.py ``` 4. **交互操作**： * 程序启动后，会提示你输入一个问题。 * 输入后，程序会自动执行搜索、抓取预设RSS源（示例中为机器之心）、并调用智谱AI。 * 稍等片刻，生成的文章就会显示在屏幕上，并可以选择保存为文本文件。 ### **五、 进阶优化与扩展** 此基础工作流已可运行，你还可以根据需求进行扩展： * **动态RSS源选择**：可以编写一个函数，从搜索结果中提取网站的域名，并尝试拼接出常见的RSS路径（如 `/feed`, `/rss`），实现RSS源的动态发现。 * **关键词优化**：在搜索前，可以先调用一次智谱AI的快速接口，对用户问题进行优化和关键词提取，以提升搜索质量 [ref_3]。 * **错误处理与重试**：为每个API调用和网络请求增加更完善的错误处理与重试机制，提升稳定性。 * **内容去重与排序**：在将信息交给AI前，对搜索和RSS获取的重复内容进行去重，并可按时间或相关性排序。 * **使用框架**：对于更复杂的生产流程，可以考虑使用如 **[Spring AI](https://spring.io/projects/spring-ai)** 这样的框架来构建和管理工具调用链 [ref_1]，或使用 **[Dify](https://dify.ai)** 等可视化工作流平台进行编排 [ref_2]。