你好，我是刘强。
本课程是关于
大模型智能体
的实战课程，包括原理、算法、应用场景、代码实战案例等，下表是本次课程的大纲。本课是第
24
节，讲解实战案例：
做一个跟踪技术热点的智能体
。本课约
6000
字，阅读时长35
min
。
以下是本次课程的正文：
上一课
我们利用LangGraph、大模型和智能体技术实现了一个个性化的语音助手，大家感受到了大模型、智能体的巨大潜力。仔细分析，个性化语音助手是一个步骤明确、有严格依赖关系的处理流，我们通过在Prompt中明确告知智能体实现步骤及对每个工具输入输出、作用进行说明，智能体能够自己确定调用的依赖关系，这已经充分体现了智能体的自主决策能力。
上面这种方法非常适合整个决策过程不是固定的场景，比如如果决策过程依赖外部的变量，这时就可以充分利用大模型的“判断力”。这种方法有一个缺点，如果智能体背后的大脑——大模型的推理能力不够强或者Prompt写的不够清晰，那么任务容易失败。那么有不有一种更高效、更稳定的实现方案呢？答案是肯定的。
针对这类流程非常明确的任务，我们可以利用另外一种方式来解决——工作流，在工作流中我们可以严格定义各个步骤之前的依赖关系，每个步骤是一个节点，它们之间的依赖关系构成一条有向边，最终整个任务可以编排成一张有向无环图（见下面图24-1）。执行引擎根据图节点、边的走向来执行，这样可以确保整个流程会更明确、不容易出错。
本课我们就基于上面的思路来实现一个自动帮我们跟踪热点技术的智能体——PaperPulse，下面我们从整体流程、工作流节点实现、工作流实现3个部分来展开讲解。
PaperPulse要解决的问题是：用户通过订阅关键词（比如agent），智能体会将最新有关agent的学术论文下载下来、然后进行翻译总结，将最终的每篇论文的摘要通过邮件发送给你。下面图24-2就是具体的流程：
本课我们用到的核心技术是LangGraph，LangGraph已经将工作流封装成了比较完整的接口了，我们只要实现每个节点，将整个工作流串联起来就好了。关于LangGraph的知识点不是本书重点，不熟悉的读者需要自己去补充相关知识（见参考文献1）。
在讲解之前，我们将代码中会用到的参数放到了配置文件中，配置文件在github仓库中对应
llm_agent_abc/automated_article/configs/model_config.py。下面的代码会会用到，后面不再说明。
DEEPSEEK_API_KEY = "你的deepseek API key"DEEPSEEK_BASE_URL = "https://api.deepseek.com/v1"DEEPSEEK_MODEL = "deepseek-chat"ARXIV_BASE_URL = "https://arxiv.org/search/advanced" SENDER_EMAIL = "[email protected]" SENDER_NAME = "Databri AI" SENDER_PASSWD = "xxx"
从图24-2我们可以看到，PaperPulse还是非常简单的，只有4个步骤，每个步骤做的事情也比较单一，比较复杂的是第三步——对Paper进行翻译，这个过程会用到大模型。下面我们对各个步骤展开说明。
第一步是从arxiv（参考文献2）基于关键词和日期来搜索最新的相关论文。下面代码中的函数
get_article_for_keywords会实现这个功能，这个函数的入参是dict，包含2个字段：keywords、days。keywords是一个list，是你想搜索的文章的关键词，我们会从论文的标题、摘要中进行匹配。days是你想搜索最近几天的文章，本课我们默认的值是1，也就是过去一天的最新文章。具体实现细节见下面代码，这里不展开讲解（本代码对应github仓库中的llm_agent_abc/automated_article/utils/get_new_articles_from_arxiv.py）。
import requestsfrom bs4 import BeautifulSoupfrom typing import Dict, Listfrom datetime import datetime, timedeltaimport syssys.path.append('./')from automated_article.configs.model_config import ARXIV_BASE_URL
def search_arxiv_advanced(base_url, params): """ 使用 arXiv 高级搜索功能获取文章信息。
:param base_url: 高级搜索 URL :param params: 搜索参数（包含关键词、日期范围等） :return: 包含文章信息的列表 """ # 请求页面 response = requests.get(base_url, params=params) if response.status_code != 200: raise Exception(f"Failed to fetch data from arXiv: {response.status_code}")
# 解析 HTML 内容 soup = BeautifulSoup(response.text, 'html.parser') results = []
# 查找搜索结果列表 articles = soup.find_all('li', class_='arxiv-result') for article in articles: title_tag = article.find('p', class_='title is-5 mathjax') abstract_tag = article.find('span', class_='abstract-full has-text-grey-dark mathjax') authors_tag = article.find('p', class_='authors') date_tag = article.find('p', class_='is-size-7')
title = title_tag.text.strip() if title_tag else "N/A" abstract = abstract_tag.text.strip() if abstract_tag else "N/A" authors = authors_tag.text.strip() if authors_tag else "N/A" date = date_tag.text.strip() if date_tag else "N/A" # 提取文章链接 link_tag = article.find('p', class_='list-title is-inline-block') link = link_tag.find('a')['href'] if link_tag and link_tag.find('a') else "N/A" results.append({ "title": title, "abstract": abstract, "authors": authors, "date": date, "link": link }) return results
def get_article_for_keywords(info: Dict) -> List[Dict]: keywords = info['keywords'] days = info['days'] # 获取昨天日期 yesterday = datetime.now() - timedelta(days=days) # 格式化为字符串 formatted_yesterday = yesterday.strftime("%Y-%m-%d") # 获取当天日期 today = datetime.now() # 格式化为字符串 formatted_today = today.strftime("%Y-%m-%d") # 输出格式：2024-11-26
start_time = info.get('start_time', formatted_yesterday) end_time = info.get('end_time', formatted_today) size = info.get('size', 25) if not keywords: return [] articles = [] for keyword in keywords: params = { "advanced": "", "terms-0-operator": "AND", "terms-0-term": keyword, "terms-0-field": "title", "terms-1-operator": "AND", "terms-1-term": keyword, "terms-1-field": "abstract", "classification-computer_science": "y", "classification-include_cross_list": "include", "date-filter_by": "date_range", "date-from_date": start_time, "date-to_date": end_time, "date-date_type": "submitted_date", "abstracts": "show", "size": size, "order": "-announced_date_first" } articles.extend(search_arxiv_advanced(ARXIV_BASE_URL, params))
if articles: print(f"With keywords {keywords}, Found {len(articles)} articles.") return articles else: return []
if __name__ == "__main__": info = { "keywords": ["agent"], "days": 1 } articles = get_article_for_keywords(info) print(f"Found {len(articles)} articles:") for article in articles: print(f"- Title: {article['title']}") print(f" Authors: {article['authors']}") print(f" Abstract: {article['abstract'][:200]}...") # 截断摘要方便查看 print(f" Link: {article['link']}") print(f" Date: {article['date']}\n")
上面第一步会获取文章对应的标题、作者、摘要、链接、日期等信息，第二步就是利用链接将这篇文章下载到本地。下面代码中的
download_all_papers函数会一次性将所有相关文章下载到本地。这一步骤的代码对应github仓库中的llm_agent_abc/automated_article/utils/download_new_articles.py。
import osimport requestsfrom tqdm import tqdm
def download_arxiv_paper(link, output_dir="./automated_article/output/articles"): """ 下载 arXiv 文章的 PDF 文件到指定目录。 :param link: arXiv 的文章链接 (如 https://arxiv.org/abs/2411.13543) :param output_dir: 保存 PDF 文件的目录 """ os.makedirs(output_dir, exist_ok=True) pdf_url = link.replace("abs", "pdf") arxiv_id = link.split("/")[-1] output_path = os.path.join(output_dir, f"{arxiv_id}.pdf") print(output_path) if os.path.exists(output_path): print(f"{arxiv_id} already exists. Skipping download.") return None else: try: print(f"Downloading {arxiv_id} from {pdf_url}...") response = requests.get(pdf_url, stream=True) response.raise_for_status()
with open(output_path, "wb") as pdf_file: for chunk in response.iter_content(chunk_size=1024): if chunk: pdf_file.write(chunk) print(f"Downloaded: {output_path}") return output_path except requests.exceptions.RequestException as e: print(f"Failed to download {arxiv_id}: {e}") return None
def download_all_papers(links: list[str], output_dir="./automated_article/output/articles"): """ 下载所有给定链接对应的 arXiv 文章。
:param links: arXiv 文章链接列表 :param output_dir: 保存 PDF 文件的目录 """ for link in tqdm(links, desc="Downloading papers"): download_arxiv_paper(link, output_dir)
if __name__ == "__main__": paper_links = [ "https://arxiv.org/abs/2411.14214", "https://arxiv.org/abs/2411.13768", "https://arxiv.org/abs/2411.13543", ] download_all_papers(paper_links, output_dir="./automated_article/output/articles")
下载好了论文，第三步就是利用大模型对论文进行总结了。这里我们会让大模型帮我们实现2个功能：一是帮我们的文章起一个吸引人的标题；二是针对文章的正文，帮我们生成一篇2000-5000字的摘要。这2个功能我们都是通过调用商业大模型（笔者是使用的DeepSeek大模型）的API通过prompt学习来实现的。具体过程读者可以看下面的代码（对应github仓库中的
llm_agent_abc/automated_article/utils/write_abstracts.py
），代码中我们用一个函数
write_abstract_api将这一步封装到了一起，可以为所有文章生成标题和摘要
。生成的摘要也会保存到本地。
import osimport PyPDF2from tqdm import tqdmimport syssys.path.append('./')from openai import OpenAIfrom automated_article.configs.model_config import DEEPSEEK_API_KEY, DEEPSEEK_MODEL, DEEPSEEK_BASE_URL
client = OpenAI( api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL,)
def generate_title(article_content): """ 使用大模型生成吸引眼球的中文标题。 :param article_content: 英文的文章内容 :return: 生成的中文标题 """ prompt = ( "你是一名专业的编辑，擅长为长篇文章撰写吸引人的中文标题。" "标题需要满足以下要求：\n" "1. 最好是疑问句，或者包含数字，或者有对比性。\n" "2. 用中文撰写，标题需要生动、有趣、简洁。\n" "3. 标题长度为10-25个字之间。\n" "4. 你只需要生成一个标题。\n" "\n" "以下是文章内容摘要：\n" f"{article_content[:3000]}\n" "\n请基于文章内容撰写一个中文标题。" ) response = client.chat.completions.create( model=DEEPSEEK_MODEL, messages=[ {"role": "system", "content": "你是一名专业的中文编辑。"}, {"role": "user", "content": prompt}, ] ) message_content = response.choices[0].message.content.strip() return message_content
def generate_summary(title, article_content): """ 使用 OpenAI 的 GPT 生成文章的中文长文总结（2000-5000字）。
:param article_content: 英文的文章内容 :return: 中文长文总结 """ prompt = ( "你是一名专业的技术作家，擅长将英文文章总结为中文长文。" "请将以下文章总结成一篇2000-5000字的中文文章，重点突出文章的核心贡献、创新方法及主要结论，并给出数据支撑（如果有的话），但不要提供参考文献。\n" "你的总结要确保语言生动有趣，不要太学术，适合大众阅读。\n" f"总结的标题为：{title}，请用这个标题，别自己起标题。\n" f"确保输出中的标题为markdown的二级标题，即标题以 ## 开头，其他子标题依次递进，分别用三级、四级、五级等标题样式。\n" f"给你提供的文章内容是：{article_content[:10000]}\n" "\n现在请你撰写总结。" ) response = client.chat.completions.create( model=DEEPSEEK_MODEL, messages=[ {"role": "system", "content": "你是一名专业的中文技术作家。"}, {"role": "user", "content": prompt}, ], max_tokens=4096 ) message_content = response.choices[0].message.content.strip() return message_content
def process_papers(file_info, file_dir: str = "./automated_article/output/articles/", output_dir: str = "./automated_article/output/summaries"): """ 遍历指定目录中的所有 PDF 文件，生成标题和总结，并保存为文本文件。
:param file_info: [{"file": xx.pdf, "link": xx}] :param file_dir: :param output_dir: """ os.makedirs(output_dir, exist_ok=True) summary_list = [] for info_ in tqdm(file_info, desc="Processing papers"): file = info_['file'] link = info_['link'] if not file.endswith(".pdf"): continue with open(os.path.join(file_dir, file), "rb") as f: reader = PyPDF2.PdfReader(f) article_content = "\n".join(page.extract_text() for page in reader.pages) title = generate_title(article_content)
summary = generate_summary(title, article_content) output_path = os.path.join(output_dir, f"{title}.md") with open(output_path, "w", encoding="utf-8") as out_file: out_file.write(f"{summary}\n")
summary_list.append({ "title": title, "summary": summary, "link": link }) return summary_list
def write_abstract_api(file_info, file_dir: str = "./automated_article/output/articles/", output_dir: str = "./automated_article/output/summaries"): summary_list = process_papers(file_info, file_dir, output_dir) return summary_list
有了文章的摘要，下面就是通过发送邮件将文章及摘要发送给指定的邮箱了，这个过程代码比较简单，你可以参考下面代码（对应github仓库中的
llm_agent_abc/automated_article/utils/send_email.py
）。我们在下一节工作流中会将所有文章合并成markdown的格式来形成邮件的正文，这里发送的就是markdown格式的邮件正文。
