Primus AI Researcher 是一个以免费方式提供的 AI 研究助手。这类工具的价值不在于“能回答”而在于把用户提出的一个问题拆解成检索、阅读、归纳、引用四个环节并最终输出一份带来源可追溯的研究结论。免费模式降低了使用门槛但同时也带来上下文长度、请求频率、可用外部服务和 credits 数量等实际约束。如果把这些约束当成无关紧要的背景知识很容易在设计阶段就把研究任务压进一个不合理的大 Prompt 里后续每一步都会被动。这篇文章不会把 Primus AI Researcher 当作黑盒来介绍而是以这类免费研究助手为参照从工程实践角度拆解它背后的模块并实现一个最小可运行的本地研究流水线。读者可以用免费额度跑通整个过程也可以把模型接口替换成本地模型或团队已有的服务。完成这套案例后你会掌握查询生成、网页抓取、文档切片、摘要生成和引用管理是如何配合的也能在后续项目中复用同一套设计。1. 先理解 AI 研究助手的工作流程再动手写代码1.1 它解决的问题不是“搜索”而是“把搜索变成可引用的结论”搜索引擎返回的是链接列表用户还需要逐条打开、阅读、判断可信度、交叉对比最后才能形成自己的结论。AI 研究助手把这段人工劳作部分自动化了。它把“搜索 - 阅读 - 得出结论”的链路缩短让用户可以更快得到一份相对完整的材料。但这里有一个容易误解的地方AI 研究助手并不代替用户做判断它只是把“找到相关信息”和“形成可读摘要”这两件事前置了。模型可能产生幻觉引用可能张冠李戴搜索结果可能来自低质量页面。因此工程上必须围绕“可验证”来设计而不是一味追求回答流畅。核心指标不是“结论多漂亮”而是“每条结论是否有来源、来源是否真实、来源是否真的支持这条结论”。1.2 标准流水线查询生成、检索、读取、归纳、引用一个典型的 AI 研究助手流水线如下查询生成用户输入问题后LLM 把问题扩展成多个角度的搜索关键词。检索对每个关键词执行搜索收集候选网页或文档。读取抓取候选页面提取正文信息去掉导航、广告、脚本等噪声。归纳把正文切片逐片总结再汇总成模块化结论。引用为每条结论绑定来源 URL 和原文片段。用伪代码可以这样表示async def research(question: str) - Report: queries await generate_queries(question) results await search(queries) docs await fetch_and_extract(results) chunks split_docs(docs) summary await summarize(chunks, question) report build_report(question, summary, chunks) return report这里要注意流水线中的每一步都可能失败。查询生成失败可以重试单个网页抓取失败不应该终止整个任务某个来源内容过短或不相关应该在归纳阶段过滤掉。真实的工程实现中这些异常分支的设计优先级甚至高于正常路径。1.3 免费模式的关键约束上下文长度、请求频率、credits很多 AI 平台使用 credits 来计量 API 调用量。免费版通常赠送一定数量的 credits而不是无限 token。理解 credits 的意义在于设计流水线时要像记账一样估算每次任务的消耗。一次典型研究任务会经历生成查询词调用一次模型每个网页切片总结可能调用几次模型最终合并摘要再调用一次模型。如果用户一次研究抓取 10 个网页每个网页切出 5 个片段那么仅片段总结就会产生 50 次左右模型调用。免费额度很可能撑不住这种消耗。所以流水线设计上要主动控制文本长度不要一次性把所有网页塞进一个 Prompt而应该先切片、再总结最后只把中间摘要提交给最终合并步骤。这也是后面实现中采用“分层总结”的原因。生产环境还要考虑请求频率限制调用模型前先评估每月或每日配额在代码里记录实际 token 消耗方便后续做成本分析。2. 环境准备和依赖安装先确认运行条件2.1 运行环境要求下面的示例基于 Python因为异步 HTTP、网页解析和文本处理生态比较成熟。如果你使用的是 Java、Go 或 Node.js思路仍然适用只是依赖库和编码方式不同。项目建议要求说明Python3.10 及以上使用 async/await 和类型注解操作系统Windows 10 / macOS / Linux主要差异在虚拟环境激活命令LLM 服务OpenAI 兼容接口或本地模型免费额度需要自行确认网络可访问公开网页和搜索 API抓取时需遵守目标站点的规则依赖管理venv 或 conda避免污染系统 Python这里使用“OpenAI 兼容接口”这个说法是因为很多模型服务商都提供类似的接口格式。具体使用哪家、模型名称是什么以你自己申请到的服务为准。2.2 安装项目依赖创建一个项目目录并初始化虚拟环境mkdir primus-researcher-demo cd primus-researcher-demo python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate然后安装依赖pip install -U httpx beautifulsoup4 trafilatura pydantic markdownify python-dotenv各个依赖的用途如下依赖用途httpx异步 HTTP 请求调用 LLM API 和抓取网页beautifulsoup4HTML 解析提取链接和正文标签trafilatura从网页中提取正文对新闻和博客类页面效果好pydantic定义数据模型校验返回的 JSON 结构markdownify把 HTML 转成 Markdown方便生成报告python-dotenv从 .env 文件加载配置这些库都是常规依赖安装过程如果较慢可以考虑使用国内镜像源但这里不展开讲换源细节。2.3 配置 LLM 服务和搜索服务在项目根目录创建.env文件LLM_BASE_URLhttps://api.example.com/v1 LLM_API_KEYyour_api_key LLM_MODELyour-model-name SEARCH_API_KEYyour_search_api_key SEARCH_ENGINE_IDyour_search_engine_id如果你使用本地模型比如已经安装了 Ollama可以把配置改成类似这样LLM_BASE_URLhttp://localhost:11434/v1 LLM_API_KEYollama LLM_MODELqwen2.5:7b这里只是示例Ollama 版本、模型名称都按本机实际情况调整。关键是理解把 LLM 访问封装成一个通用客户端之后切换在线服务和本地模型只需要改配置不需要改业务代码。3. 实现一个最小可运行的 AI 研究流水线3.1 项目结构设计目录结构如下primus-researcher-demo/ ├── .env ├── requirements.txt ├── primus/ │ ├── __init__.py │ ├── config.py │ ├── llm.py │ ├── search.py │ ├── extract.py │ ├── chunk.py │ ├── summarize.py │ └── report.py └── main.py每个文件的职责很明确config.py 负责读取配置llm.py 封装模型调用search.py 负责查询生成和搜索extract.py 负责抓取网页并提取正文chunk.py 负责文本切片summarize.py 负责分层总结report.py 负责生成最终报告main.py 组装整个流程。3.2 配置模块统一管理 API Key 和参数import os from pydantic import BaseModel from dotenv import load_dotenv load_dotenv() class Settings(BaseModel): llm_base_url: str os.getenv(LLM_BASE_URL, https://api.example.com/v1) llm_api_key: str os.getenv(LLM_API_KEY, ) llm_model: str os.getenv(LLM_MODEL, gpt-4o-mini) search_api_key: str os.getenv(SEARCH_API_KEY, ) search_engine_id: str os.getenv(SEARCH_ENGINE_ID, ) max_chunk_size: int 1200 max_chunks_per_doc: int 20 settings Settings()使用 pydantic 的好处是配置字段有类型约束万一漏了必填项启动阶段就能发现。API Key 不要写死在代码里统一走环境变量或密钥管理服务。3.3 LLM 客户端用统一接口隔离不同模型服务import httpx class LLMClient: def __init__(self, base_url: str, api_key: str, model: str): self.base_url base_url.rstrip(/) self.api_key api_key self.model model self.headers { Authorization: fBearer {api_key}, Content-Type: application/json, } async def chat( self, messages: list[dict], temperature: float 0.2, max_tokens: int 1024, ) - str: payload { model: self.model, messages: messages, temperature: temperature, max_tokens: max_tokens, } async with httpx.AsyncClient(timeout60) as client: resp await client.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content]这里使用 OpenAI 兼容的/chat/completions协议方便在不同服务商之间切换。实际服务商可能使用不同的请求字段例如max_completion_tokens落地前要以官方 SDK 和文档为准。这个封装只负责“发送消息并返回文本”不处理业务逻辑后续替换成本地模型时只改这一层。3.4 查询生成让模型把用户问题拆成多个搜索词用户的问题是宽泛的直接用原问题去搜索通常只能得到单一角度的结果。查询生成模块让模型拆解出多个独立搜索词import json async def generate_queries(llm: LLMClient, question: str) - list[str]: prompt f你是一个研究助理。请把下面的问题拆解成 3 到 5 个独立的搜索查询。 每个查询只保留关键词和必要的关系词不要出现完整长句。 只输出 JSON 数组不要输出额外内容。 问题{question} messages [ {role: system, content: 你是一名严谨的信息检索助手。}, {role: user, content: prompt}, ] text await llm.chat(messages, temperature0.1, max_tokens512) # 简易解析生产环境建议增加异常处理 return json.loads(text)这个模块的产出直接影响后续检索质量。如果提示词允许模型输出解释性文字解析 JSON 时会报错。所以提示词里要明确“只输出 JSON 数组”。如果模型返回格式不稳定可以在解析失败时让模型重试一次。3.5 搜索执行把查询词提交到搜索服务不同搜索服务的 API 差异很大。下面以 Google Custom Search JSON API 为例展示标准的请求包装方式async def search_web( query: str, api_key: str, engine_id: str, num: int 5, ) - list[dict]: params { key: api_key, cx: engine_id, q: query, num: min(num, 10), } async with httpx.AsyncClient(timeout30) as client: resp await client.get( https://www.googleapis.com/customsearch/v1, paramsparams, ) resp.raise_for_status() items resp.json().get(items, []) results [] for item in items: results.append( { title: item.get(title, ), url: item.get(link, ), snippet: item.get(snippet, ), } ) return results如果你没有搜索 API 的免费额度也可以退化为一个“种子 URL 列表”由用户在配置里预先指定一批候选网页。真实项目中搜索服务经常需要二次过滤例如排除某些域名、排除明显低质量的目录页。3.6 正文提取去掉导航和广告只留正文搜索返回的 URL 只是第一步更重要的是从网页中提取正文。trafilatura 是这一环节比较成熟的库import asyncio import trafilatura async def fetch_and_extract(url: str, timeout: float 15.0) - dict: try: downloaded await asyncio.to_thread(trafilatura.fetch_url, url) if not downloaded: return {url: url, content: , error: fetch empty} text await asyncio.to_thread(trafilatura.extract, downloaded) if not text or len(text.strip()) 100: return {url: url, content: , error: extract too short} return {url: url, content: text, error: } except Exception as exc: return {url: url, content: , error: str(exc)}这里用asyncio.to_thread把同步的 trafilatura 调用放到线程池避免阻塞事件循环。抓取失败时返回错误信息而不是抛异常这样主流程可以继续处理其他网页。注意抓取公开网页时要遵守目标站点的 robots.txt 和服务条款控制请求频率。这里所有抓取仅用于学习和技术验证不要对单个站点做高并发抓取。3.7 文档切片控制每次提交给模型的文本长度网页正文可能很长直接塞进模型会超过上下文限制也可能产生较高费用。因此需要把文本切片def chunk_text( text: str, max_chunk_size: int 1200, overlap: int 100, ) - list[str]: paragraphs [p.strip() for p in text.split(\n) if p.strip()] chunks [] current for para in paragraphs: if len(current) len(para) 2 max_chunk_size: if current: chunks.append(current) current current[-overlap:] if overlap else current current \n para if current else para if current: chunks.append(current) return chunks这里的切片策略是“按段落累积超出阈值则截断并保留末尾一小段作为重叠”。重叠的作用是缓解段落边界处的语义断裂。如果文本是结构化很强的文档更好的做法是按标题层级切分也就是 Markdown Heading 感知切片。3.8 分层总结先总结片段再合并成结论文案不要试图一次把所有文本归纳完。先让 LLM 对每个切片做小结再把多个小结合并成最终结论async def summarize_chunk( llm: LLMClient, chunk: str, question: str, ) - str: messages [ { role: system, content: 你是研究助理只根据给出的文本做客观摘要不要补充外部知识。, }, { role: user, content: f问题{question}\n\n文本片段\n{chunk}\n\n请给出 3 到 5 条要点每条不超过 50 字。, }, ] return await llm.chat(messages, temperature0.2, max_tokens600) async def merge_summaries( llm: LLMClient, summaries: list[str], question: str, ) - str: joined \n.join(f- {s} for s in summaries) messages [ { role: system, content: 你是研究助理负责合并多个片段摘要输出结构化的最终结论。, }, { role: user, content: f问题{question}\n\n片段摘要\n{joined}\n\n请输出1. 核心结论2. 分点说明3. 仍不确定的地方。, }, ] return await llm.chat(messages, temperature0.2, max_tokens1500)这个设计把成本分散到多个小请求中。局部总结即使丢失少量信息也只影响单个片段但如果用一个超大 Prompt很可能直接超过上下文限制导致整个任务失败。3.9 主流程组装import asyncio from primus.config import settings from primus.llm import LLMClient from primus.search import generate_queries, search_web from primus.extract import fetch_and_extract from primus.chunk import chunk_text from primus.summarize import summarize_chunk, merge_summaries async def research(question: str): llm LLMClient( settings.llm_base_url, settings.llm_api_key, settings.llm_model, ) queries await generate_queries(llm, question) print(查询词, queries) all_results [] for q in queries: try: results await search_web( q, settings.search_api_key, settings.search_engine_id, ) all_results.extend(results) except Exception as exc: print(f搜索失败 {q}: {exc}) seen set() unique_results [] for r in all_results: if r[url] not in seen: seen.add(r[url]) unique_results.append(r) docs [] for r in unique_results[:10]: doc await fetch_and_extract(r[url]) if doc[content]: docs.append(doc) chunk_summaries [] for doc in docs: chunks chunk_text(doc[content], settings.max_chunk_size) for chunk in chunks[: settings.max_chunks_per_doc]: try: summary await summarize_chunk(llm, chunk, question) chunk_summaries.append(summary) except Exception as exc: print(f片段总结失败 {doc[url]}: {exc}) final await merge_summaries(llm, chunk_summaries, question) print( 最终结论 ) print(final) print( 来源链接 ) for doc in docs: print(f- {doc[url]}) if __name__ __main__: asyncio.run(research(为什么大型语言模型需要 RAG 而不是单独微调))这个主流程有几个刻意安排搜索失败和片段总结失败都做了 try except不会让单点错误拖垮整个任务。抓取结果去重避免同一个 URL 被多条查询词重复提交。限制最多处理 10 个网页每个网页最多切 20 个片段控制免费额度消耗。最终输出“结论 来源链接”两部分至少让用户可以核对结论来自哪些页面。4. 运行验证与结果分析4.1 运行命令配置好.env后直接运行python main.py预期输出顺序如下查询词列表通常是 3 到 5 个字符串。每个网页抓取时的处理日志。最终结论包含核心结论、分点说明和不确定项。来源链接列表。如果看到查询词数量明显偏少或者最终结论里没有分点需要检查模型输出是否符合提示词约束。4.2 如何判断摘要没有失真不要只看输出是否流畅要从三个角度验证检查项判断标准覆盖度问题的多个角度是否都被讨论到忠实度结论里的关键数据、名词、时间是否能在来源页面中找到可读性分点说明是否逻辑清晰而不是简单的要点堆砌最直接的办法是随机挑一条结论去对应来源链接里搜索结论中的关键短语。如果原文根本不存在这个说法说明模型发生了幻觉需要收紧“只根据文本总结”的提示词或者在切片阶段增加更严格的文档相关性过滤。4.3 引用一致性检查当前实现把“来源链接”放在最后列出并没有做到“每条结论对应一条来源”。严格的生产实现应该让每个片段带上 doc_id并且在总结时保留这个 doc_id。可以这样改进# chunk 阶段为每个片段绑定文档 ID doc_chunks [] for doc_id, doc in enumerate(docs): for chunk in chunk_text(doc[content], settings.max_chunk_size): doc_chunks.append({doc_id: doc_id, chunk: chunk}) # 总结时要求模型输出 JSON每个要点都带上 doc_id # [{point: ..., doc_id: 0}]最终报告输出时用 doc_id 映射回 URL这样每条结论都能追溯到具体来源。也可以在报告里额外输出“该结论支持的证据原文片段”方便读者验证。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。一次运行只能证明“链路通”不能证明“结论可靠”。5. 常见问题排查5.1 API 返回 401 或 403现象调用 LLM 或搜索服务时返回认证失败。可能原因API Key 配置错误、Key 已失效、服务商不支持当前请求头。检查方式# 查看 .env 是否加载成功 python -c from primus.config import settings; print(settings.llm_base_url); print(settings.llm_api_key[:4])处理建议确认 Authorization 头格式是否与服务商文档一致。部分服务使用Bearer部分使用自定义头。如果密钥含特殊字符检查.env中是否被引号包裹。5.2 API 返回 429 或请求被限流现象前面几次调用正常跑到中途突然收到限流错误。可能原因免费额度已用完或者短时间请求频率超过限制。检查方式查看响应头里的Retry-After字段统计当前任务已经发起了多少次模型调用。处理建议在LLMClient.chat中增加退避重试逻辑。最简单的方式是捕获 429 后等待 2 的指数次幂秒再重试但重试次数不要无限增加。更根本的办法是减少文档数量、减少切片数量或者使用更小的模型做局部总结。5.3 正文提取为空现象搜索返回了 URL但fetch_and_extract返回的 content 是空字符串。可能原因目标站点禁止爬虫、页面由 JavaScript 动态渲染、页面内容太短、正文提取库识别失败。检查方式用浏览器打开该 URL确认页面是否有正文文本。然后用命令行单独测试import trafilatura downloaded trafilatura.fetch_url(https://example.com/page) print(downloaded[:500]) print(trafilatura.extract(downloaded))处理建议如果页面是动态渲染需要引入 headless 浏览器方案但成本较高免费版不建议默认开启。更稳妥的做法是过滤这些 URL设置“提取失败则跳过”不要让单个站点拖慢整个任务。5.4 总结内容与原文不符现象结论说法流畅但去来源页面里找不到对应论据。可能原因切片过碎导致上下文不足提示词没有强调“只根据文本”模型在拼接多个片段时产生了自己的推理。检查方式把对应 chunk 单独拿出来重新调用一次summarize_chunk看是否仍然出现同样说法。处理建议增加切片的 overlap提示词中明确要求“如果文本中没有提到不要补充”。更严格的做法是让模型输出要点时直接引用原文中的短语报告生成时展示原文引用片段而不是让模型自由表述。5.5 引用链接失效或来源张冠李戴现象生成的报告里每条结论都带了链接但链接打不开或者链接里的内容和结论完全无关。可能原因搜索返回过期链接网页在抓取后发生变化doc_id 在分片过程中错位某个网页正文被解析成了另一篇文章的内容。检查方式记录每个 doc_id 对应的 URL以及该 doc_id 输出的所有 chunk逐个核对。处理建议报告生成前对 URL 做状态码检查过滤 404 和登录墙。在 chunk 绑定 doc_id 时打印日志方便反查。问题现象可能原因检查方式处理建议API 返回 401API Key 无效或格式错误打印请求头和 key 前几位重新生成 key确认认证头格式API 返回 429请求频率超限或额度用完查看 Retry-After、统计调用次数退避重试减少文档数量查询词为空模型返回 JSON 解析失败打印模型原始响应增加异常重试收紧提示词正文提取为空反爬、动态渲染、页面过短浏览器打开 URL单测 trafilatura过滤失败 URL必要时换源总结与原文不符切片过碎、提示词约束不足单测对应 chunk 的总结结果增加 overlap要求引用原文引用链接失效过期链接、登录墙检查 HTTP 状态码报告生成前过滤无效链接6. 从本地脚本到生产级研究服务6.1 学习环境只需要串通流程在学习阶段一个main.py足够。不需要引入消息队列、缓存和监控打印日志就能定位问题。这时的目标不是并行处理几十个问题而是把“查询生成、搜索、抓取、切片、总结、引用”这条链路完整体验一遍。推荐做法是先用一个自己熟悉的问题跑通然后故意制造失败场景比如提供一个会 404 的 URL、让 API Key 失效一次、提交一个超过上下文长度的文本观察系统如何表现。这些边界测试比顺顺利利跑通一次更能帮助你理解模块边界。6.2 生产环境必须补上的能力生产环境不是再把print换成日志那么简单至少要补齐以下模块能力说明配置外置和密钥管理API Key 走密钥服务不能进代码仓库任务队列使用 Celery、Redis Stream 或云上的异步任务服务缓存相同问题、相同 URL 的抓取结果要缓存减少重复调用限流和退避对不同 API 设置独立的速率限制策略结构化日志记录每次任务的 query、耗时、token 消耗、失败 URL监控告警设置成功率、平均耗时、credits 消耗趋势告警结果去重对同一 URL 在不同查询词下的重复出现做去重引用校验生成报告前验证 URL 可访问、结论能在原文中找到依据数据留存报告版本、原始抓取内容、中间摘要要留存便于审计异常降级单站点抓取失败时跳过模型调用失败时重试免费工具通常只能在成本和稳定性之间做取舍。如果你要在生产环境长期运行建议从一开始就记录 token 消耗和 credits 消耗否则账单容易失控。6.3 credits 消耗控制策略一次研究任务的消耗不是固定值取决于文档数量和切片数量。可以这样估算假设每个切片约 1200 字片段总结请求约 900 tokens10 个文档每个切 5 个片段片段总结阶段就约消耗 45000 tokens最终合并摘要还要再加 2000 tokens 左右。控制策略通常包括限制抓取文档数量例如最多 10 个。限制每个文档切片数量例如最多 20 个。局部总结使用较小的模型最终合并使用更强的模型。对正文按长度排序优先总结与查询词相关性高的页面。对相同 URL 的抓取结果做持久化缓存减少重复抓取。免费额度是一个动态数字不要在任何代码里写死某个固定额度值。更合理的做法是把剩余 credits 作为配置项在任务开始前检查剩余量低于阈值就停止新任务。6.4 从脚本到 Agent研究助手的更完整形态是 AI Agent模型可以决定调用哪些工具、是否继续追问、是否重新搜索而不再是一次性的五步流水线。在这个方向下前面实现的模块都可以改造成工具函数由 Agent 根据用户问题选择调用。如果你所在团队的主力技术栈是 Java可以关注 Spring AI 这类生态组件把研究助手的部分能力织入 Spring Boot 服务。不过框架版本迭代较快落地前要以对应版本的官方文档为准。对个人开发者来说先用 Python 跑通这套流水线、理解数据流再迁移到其他技术栈会容易得多。7. 最佳实践和扩展方向7.1 可复用检查清单无论你搭建的是免费研究工具还是付费服务下面的清单都可以在发布前过一遍阶段检查项输入问题是否明确是否需要限定时间范围和地域查询生成查询词是否覆盖多个角度是否包含必要关键词检索是否去重是否过滤明显低质量页面抓取是否遵守 robots.txt是否设置超时正文提取是否过滤导航、广告、登录墙页面切片chunk 大小是否匹配模型上下文overlap 是否合理局部总结是否只依据文本是否要求模型输出要点最终合并是否区分“核心结论”和“不确定项”引用每条结论是否可追溯到具体来源输出报告是否结构化来源链接是否完整日志是否记录了每次任务的耗时、token 消耗和失败原因7.2 可以继续深入的方向这套最小流水线可以往多个方向扩展RAG把历史研究报告和常用资料存入向量数据库检索阶段先查库再搜索降低对实时搜索的依赖。多轮追问保存对话状态让用户基于上一份报告继续深入某个子问题。多主题并行研究用任务队列把多个问题分发到不同工作进程再汇总结果。报告版本管理每次研究报告生成后存入对象存储方便回溯。评估集准备一批有标准答案的问题集定期评估总结准确率而不是凭感觉判断模型好坏。工具级 Agent把生成查询、搜索、抓取、总结都封装成工具允许模型自主决定执行顺序。7.3 AI 编程辅助下的实现建议如果你在搭建过程中使用了 AI 编程工具建议先明确数据模型和接口边界再让 AI 生成具体代码。上面实现里的LLMClient.chat、search_web