基于GPT与必应搜索的RAG系统：实现AI联网问答的工程实践

1. 项目概述与核心价值最近在折腾AI应用开发的朋友估计都绕不开一个核心需求如何让大语言模型LLM不再“一本正经地胡说八道”而是能基于真实、实时的信息来回答问题。这正是“bujnlc8/gptbing”这个项目试图解决的痛点。简单来说它是一个将GPT系列模型与必应Bing搜索引擎能力深度集成的开源工具。它的核心价值在于通过一个相对轻量级的架构赋予GPT模型“联网搜索”和“事实核查”的能力从而生成更准确、更具时效性的回答。我自己在开发一些需要实时信息支持的AI助手时就深受“模型知识截止日期”的困扰。比如你问它“今天某支股票表现如何”或者“刚刚发布的某款手机参数是什么”基于静态数据训练的模型要么答非所问要么直接告诉你它不知道。而“gptbing”这类项目的思路就是让模型学会“不懂就问”——当它识别到用户的问题需要最新信息时自动调用必应搜索接口获取相关网页摘要然后将这些摘要作为上下文喂给模型让模型基于这些新鲜“食材”来“烹饪”出最终答案。这不仅仅是接个API那么简单它涉及到查询理解、搜索策略、结果筛选、上下文构建和提示词工程等一系列环节的精细打磨。这个项目特别适合两类开发者一是希望为自己的AI应用快速增加联网搜索功能的个人或小团队可以基于此进行二次开发二是对AI Agent智能体或RAG检索增强生成技术感兴趣的实践者可以将其作为一个绝佳的学习案例理解如何将外部工具搜索引擎与LLM进行有效协同。接下来我将从设计思路、核心实现、实操部署到避坑经验为你完整拆解这个项目。2. 整体架构与设计思路拆解2.1 核心工作流解析“gptbing”项目的核心工作流是一个典型的“感知-决策-执行”循环模拟了一个拥有搜索工具的智能助手的行为逻辑。其工作流可以清晰地分为以下几个阶段问题分析与意图识别当用户输入一个问题后系统首先会判断这个问题是否需要借助外部实时信息来回答。这一步至关重要避免了不必要的搜索开销例如询问“11等于几”这种常识性问题。项目通常会通过一个精心设计的提示词Prompt让GPT模型自身来判断是否需要搜索。这个提示词会引导模型思考“仅凭我的内部知识能否准确、完整地回答这个问题这个问题是否涉及近期事件、实时数据或非常具体的细节”搜索查询生成如果模型判定需要搜索下一步就是将用户的自然语言问题转化成一个或多个高效的搜索引擎查询关键词。这里不能简单地把原问题丢给搜索引擎因为其中可能包含大量对搜索无用的修饰词。例如用户问“帮我找找看最近有没有评价比较好的、适合编程的轻薄笔记本电脑推荐”模型需要提取出核心关键词如“2024年 轻薄本 编程 评测 推荐”。这个过程同样由GPT模型在提示词的引导下完成。执行搜索与结果获取系统使用配置好的必应搜索API发送上一步生成的查询关键词并获取返回的搜索结果。通常API会返回一系列网页的标题、链接和摘要Snippet。项目需要处理API的调用频率限制、网络错误等异常情况。搜索结果处理与摘要直接返回的搜索结果摘要可能冗长、重复或质量参差不齐。项目需要对结果进行去重、排序和精炼。更高级的实现中可能会让GPT模型对多个搜索结果摘要进行二次总结和整合生成一个更凝练、信息密度更高的背景资料块。最终答案生成将原始用户问题连同处理后的搜索摘要作为上下文一起提交给GPT模型要求其基于这些实时信息生成最终答案。这里的提示词需要明确指令模型“请严格依据提供的搜索资料来回答问题如果资料中没有相关信息请如实说明不知道不要编造。”2.2 技术选型与方案考量项目的技术栈选择体现了在易用性、成本和效果之间的平衡后端框架常见的选择是FastAPI或Flask。FastAPI凭借其异步支持、自动API文档生成和更高的性能成为这类AI应用后端的热门选择。它能够更好地处理同时到来的多个搜索-生成请求。大语言模型接口项目核心依赖OpenAI的Chat Completions API对应GPT-3.5/4系列或与OpenAI API兼容的其他服务如Azure OpenAI。选择OpenAI API而非本地部署模型主要是出于开发效率和模型能力的考虑。云端API省去了复杂的模型部署、优化和硬件成本能快速获得强大的对话与文本理解能力。搜索引擎接口选择了必应搜索API。相较于其他搜索引擎API必应搜索的结果质量在中文和英文场景下都比较均衡且其API提供了结构化程度较好的返回结果包含标题、链接、摘要便于后续处理。需要注意的是必应搜索API有每日调用次数限制并且需要申请相应的订阅密钥。异步处理为了提升响应速度尤其是在“搜索”和“生成”这两个可能比较耗时的I/O操作上项目通常会采用异步编程如Python的asyncio和aiohttp。当模型在“思考”是否需要搜索时网络请求可以并行准备从而减少用户等待时间。注意使用必应搜索API和OpenAI API都涉及服务费用。必应搜索有免费额度但有限制超出需付费。OpenAI API按Token用量计费。在开发和部署时务必在代码中做好用量监控和成本控制避免意外的高额账单。3. 核心模块详解与代码实现3.1 搜索决策与查询生成模块这是整个系统的“大脑”决定了搜索的触发与否以及搜索的质量。其实现通常封装在一个独立的函数或类中。import openai from typing import Optional, List class SearchOrchestrator: def __init__(self, openai_api_key: str): openai.api_key openai_api_key # 定义决策和查询生成的系统提示词 self.decision_prompt 你是一个智能助手需要判断用户的问题是否需要通过联网搜索最新信息来回答。 请根据以下规则判断 1. 如果问题涉及**今天、昨天、本周、本月、今年**等明确的时间范围且内容可能随时间变化如新闻、股价、天气、体育赛事结果则需要搜索。 2. 如果问题涉及非常具体或小众的事实、数据、产品参数、最新发布的政策法规等而你可能不了解或知识已过时则需要搜索。 3. 如果问题是关于常识、概念解释、数学计算、编程代码不涉及最新库版本等静态知识则不需要搜索。 请只输出“YES”或“NO”不要输出其他任何内容。 用户问题{user_question} self.query_gen_prompt 根据用户的问题生成最适合用于网页搜索的1-3个关键词或短语。 要求关键词必须简洁、精准能直接用于搜索引擎。去除问题中的疑问词和修饰性语言。 只输出关键词用逗号分隔。 用户问题{user_question} async def decide_if_search_needed(self, user_question: str) - bool: 决策是否需要搜索 try: response await openai.ChatCompletion.acreate( modelgpt-3.5-turbo, # 使用成本较低的模型进行决策 messages[ {role: system, content: 你是一个严谨的判断助手。}, {role: user, content: self.decision_prompt.format(user_questionuser_question)} ], temperature0.1, # 低随机性确保输出稳定为YES/NO max_tokens10, ) decision response.choices[0].message.content.strip().upper() return decision YES except Exception as e: # 如果决策过程出错保守起见不进行搜索避免错误传播 print(f决策过程出错: {e}) return False async def generate_search_queries(self, user_question: str) - List[str]: 生成搜索查询词 try: response await openai.ChatCompletion.acreate( modelgpt-3.5-turbo, messages[ {role: system, content: 你是一个搜索关键词提取专家。}, {role: user, content: self.query_gen_prompt.format(user_questionuser_question)} ], temperature0.3, max_tokens50, ) queries_text response.choices[0].message.content.strip() # 按逗号分割并清理空白字符 queries [q.strip() for q in queries_text.split(,) if q.strip()] return queries[:3] # 最多返回3个查询词 except Exception as e: print(f生成查询词出错: {e}) # 降级方案简单返回用户问题的前几个词 return [user_question[:30]]实操心得决策提示词是关键decision_prompt的编写需要大量测试和调优。过于激进会导致所有问题都触发搜索增加成本和延迟过于保守则失去了联网搜索的意义。我建议在真实对话日志中采样一批问题手动标注是否需要搜索然后用这个数据集来迭代优化你的提示词。使用低成本模型做决策决策和查询生成不需要很强的创造力使用gpt-3.5-turbo这类更便宜的模型足以胜任可以有效降低整体API调用成本。做好错误处理网络超时、API限额、模型输出格式意外等情况必须考虑。如上代码所示在try-except块中捕获异常并提供降级方案如返回原问题片段是保证系统鲁棒性的必要手段。3.2 必应搜索与结果处理模块这个模块负责与必应API交互并对返回的原始数据进行清洗和格式化。import aiohttp import asyncio from typing import Dict, Any, List import json class BingSearchClient: def __init__(self, subscription_key: str, endpoint: str https://api.bing.microsoft.com/v7.0/search): self.subscription_key subscription_key self.endpoint endpoint self.headers {Ocp-Apim-Subscription-Key: subscription_key} async def search(self, query: str, count: int 5) - List[Dict[str, str]]: 执行单次搜索返回处理后的结果列表 params {q: query, count: count, responseFilter: Webpages, textFormat: HTML} async with aiohttp.ClientSession() as session: try: async with session.get(self.endpoint, headersself.headers, paramsparams) as response: if response.status 200: data await response.json() return self._process_search_results(data) else: print(f必应搜索API错误状态码: {response.status}) return [] except aiohttp.ClientError as e: print(f网络请求错误: {e}) return [] except asyncio.TimeoutError: print(搜索请求超时) return [] def _process_search_results(self, raw_data: Dict[str, Any]) - List[Dict[str, str]]: 处理原始JSON数据提取有用信息 processed_results [] if webPages in raw_data and value in raw_data[webPages]: for item in raw_data[webPages][value]: # 提取标题、链接、摘要snippet result { title: item.get(name, ), url: item.get(url, ), snippet: item.get(snippet, ) } # 简单的质量过滤摘要不能太短且应包含查询词的相关信息这里简化处理 if len(result[snippet]) 20: processed_results.append(result) # 去重根据URL去重 seen_urls set() unique_results [] for res in processed_results: if res[url] not in seen_urls: seen_urls.add(res[url]) unique_results.append(res) return unique_results[:5] # 确保返回不超过5个结果 async def search_multiple_queries(self, queries: List[str]) - List[Dict[str, str]]: 并发执行多个查询词的搜索 tasks [self.search(q) for q in queries] all_results await asyncio.gather(*tasks, return_exceptionsTrue) # 合并结果并再次去重 merged_results [] for result in all_results: if isinstance(result, list): merged_results.extend(result) # 简单的基于标题和摘要相似度的去重此处简化实际可使用文本相似度算法 final_results [] seen_content set() for res in merged_results: content_key f{res[title][:50]}_{res[snippet][:100]} if content_key not in seen_content: seen_content.add(content_key) final_results.append(res) return final_results[:8] # 合并后返回最多8个最相关的结果注意事项API密钥管理绝不要将subscription_key硬编码在代码中或上传到公开仓库。务必使用环境变量或配置文件来管理。频率限制与重试必应API有明确的每秒查询数QPS和每月调用量限制。在生产环境中你需要实现一个带有退避策略的重试机制例如使用tenacity库并在代码中记录调用次数接近限额时发出警报。结果质量过滤_process_search_results方法中的过滤逻辑非常基础。在实际应用中你可能需要更复杂的策略比如根据域名权威性、摘要与查询的相关性可用余弦相似度初步计算、发布时间如果API提供等进行排序和筛选。3.3 答案合成与生成模块这是最后一步也是直接面向用户输出的一步。它的任务是将搜索到的信息与用户问题结合生成一个连贯、准确、注明来源的答案。class AnswerSynthesizer: def __init__(self, openai_api_key: str): openai.api_key openai_api_key def _format_context_from_results(self, search_results: List[Dict[str, str]]) - str: 将搜索结果格式化为LLM易于理解的上下文文本 context_lines [] for i, res in enumerate(search_results): # 为每个结果编号并包含标题和摘要方便后续引用 context_lines.append(f[来源{i1}] 标题{res[title]}\n摘要{res[snippet]}\n) return \n.join(context_lines) async def generate_final_answer(self, user_question: str, search_results: List[Dict[str, str]]) - Dict[str, Any]: 基于用户问题和搜索结果生成最终答案 if not search_results: # 如果没有搜索结果则告知用户并尝试仅用模型知识回答或直接说明无法回答 no_search_prompt f用户问题{user_question}\n\n未找到相关的实时信息。请根据你的知识尝试回答如果不知道或不确定请明确说明。 final_answer, used_sources await self._call_llm(no_search_prompt, []) used_sources [] # 未使用任何来源 else: formatted_context self._format_context_from_results(search_results) synthesis_prompt f请根据以下提供的搜索资料回答用户的问题。 要求 1. 答案必须严格基于提供的资料。如果资料中没有相关信息请说“根据现有资料未找到相关信息”不要编造。 2. 答案应组织得清晰、有条理。 3. 在答案中引用资料时请使用“[来源X]”的格式X对应资料前的编号。 搜索资料 {formatted_context} 用户问题{user_question} 请开始你的回答 final_answer, used_sources await self._call_llm(synthesis_prompt, search_results) return { answer: final_answer, sources: used_sources, # 包含实际被引用的来源信息标题和URL has_search_context: bool(search_results) } async def _call_llm(self, prompt: str, all_sources: List[Dict[str, str]]) - (str, List[Dict]): 调用LLM API并解析答案中的引用 try: response await openai.ChatCompletion.acreate( modelgpt-4, # 合成答案对质量要求高建议使用更强的模型 messages[ {role: system, content: 你是一个严谨的助手总是根据提供的事实来回答问题。}, {role: user, content: prompt} ], temperature0.7, max_tokens1500, ) answer_text response.choices[0].message.content.strip() # 一个简单的来源引用解析示例匹配 [来源1], [来源2] 这样的模式 import re source_refs re.findall(r\[来源(\d)\], answer_text) used_indices set(int(idx) for idx in source_refs if idx.isdigit()) used_sources [all_sources[i-1] for i in used_indices if 0 i len(all_sources)] return answer_text, used_sources except Exception as e: print(f生成最终答案时出错: {e}) return 抱歉生成答案时出现错误。, []核心技巧提示词工程是灵魂synthesis_prompt直接决定了答案的质量。明确指令模型“严格基于资料”并“注明引用”能有效减少幻觉Hallucination。你可以通过少量示例Few-shot Learning在提示词中展示理想的回答格式。引用追踪代码中简单的正则表达式匹配只能处理最基本的引用格式。更可靠的做法是要求模型在输出时使用一种结构化的格式例如JSON将答案文本和引用来源分开。或者可以使用像LangChain这样的框架它内置了更完善的引用追溯功能。模型选择答案合成步骤建议使用能力更强的模型如GPT-4因为它需要理解多篇资料进行逻辑整合并生成流畅文本。虽然成本更高但效果提升显著。4. 完整服务集成与API暴露将上述模块组合起来并通过一个Web API如FastAPI暴露给前端或其他服务调用。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn from typing import Optional app FastAPI(titleGPTBing 智能搜索问答服务) # 初始化各个组件密钥应从环境变量读取 search_client BingSearchClient(subscription_keyos.getenv(BING_SUBSCRIPTION_KEY)) search_orchestrator SearchOrchestrator(openai_api_keyos.getenv(OPENAI_API_KEY)) answer_synthesizer AnswerSynthesizer(openai_api_keyos.getenv(OPENAI_API_KEY)) class QueryRequest(BaseModel): question: str # 可选的参数用于强制是否搜索默认由AI决策 force_search: Optional[bool] None class QueryResponse(BaseModel): answer: str sources: List[Dict[str, str]] need_search: bool search_queries_used: List[str] app.post(/ask, response_modelQueryResponse) async def ask_question(request: QueryRequest): 主问答接口 user_question request.question.strip() if not user_question: raise HTTPException(status_code400, detail问题不能为空) # 1. 决策是否需要搜索 need_search request.force_search if need_search is None: need_search await search_orchestrator.decide_if_search_needed(user_question) search_results [] search_queries_used [] # 2. 如果需要执行搜索 if need_search: search_queries await search_orchestrator.generate_search_queries(user_question) search_queries_used search_queries if search_queries: search_results await search_client.search_multiple_queries(search_queries) # 如果搜索无结果可以记录日志或调整策略 if not search_results: print(f警告针对问题‘{user_question}’的搜索未返回结果。) # 3. 生成最终答案 answer_data await answer_synthesizer.generate_final_answer(user_question, search_results) return QueryResponse( answeranswer_data[answer], sourcesanswer_data[sources], need_searchneed_search, search_queries_usedsearch_queries_used ) if __name__ __main__: # 启动服务监听本地端口 uvicorn.run(app, host0.0.0.0, port8000)部署与运行将上述代码保存为main.py。在终端设置环境变量export OPENAI_API_KEY你的OpenAI密钥 export BING_SUBSCRIPTION_KEY你的必应密钥安装依赖pip install fastapi uvicorn openai aiohttp pydantic运行服务python main.py访问http://localhost:8000/docs即可看到自动生成的API文档并进行测试。5. 常见问题、优化方向与避坑指南在实际部署和优化“gptbing”这类项目的过程中你会遇到一系列典型问题。下面是我踩过坑后总结出的经验。5.1 典型问题与排查技巧问题现象可能原因排查步骤与解决方案响应速度慢1. 网络延迟高特别是调用海外API。2. 同步阻塞代码。3. GPT模型尤其是GPT-4自身生成速度慢。4. 搜索查询词过多或结果过多处理耗时。1.使用异步确保所有网络请求OpenAI、Bing都使用异步库aiohttp。2.并行化多个搜索查询词应并发执行asyncio.gather。3.设置超时为API调用设置合理的超时时间如10秒避免单个请求卡死整个流程。4.优化搜索策略限制生成的查询词数量如最多2个限制每个查询返回的结果数如3-5个。5.考虑缓存对常见问题或搜索结果进行短期缓存减少重复调用。答案不准确或“幻觉”1. 搜索返回的资料质量差或无关。2. 提示词未强制模型基于资料回答。3. 上下文过长模型忽略了尾部资料。4. 模型能力不足。1.优化搜索查询分析生成的查询词是否精准调整查询生成提示词。2.强化提示词在合成答案的提示词中使用更强烈的指令如“你必须且只能使用以下资料”并加入“如果资料中没有请说不知道”的示例。3.精简上下文对搜索结果摘要进行压缩和总结只保留最相关的部分确保关键信息在模型上下文窗口的前部。4.升级模型尝试使用GPT-4等更强大的模型进行答案合成。必应API返回空结果或错误1. API密钥无效或过期。2. 达到调用频率或月度限额。3. 查询词触发了安全策略或被限制。4. 网络问题。1.检查密钥确认密钥正确且未过期。2.查看用量登录Azure门户查看必应搜索资源的用量和限额。3.审查查询词避免使用可能被认定为恶意或违规的查询词。4.实现重试与降级代码中实现指数退避的重试机制。对于非关键问题可以降级为不使用搜索资料直接回答。成本失控1. 决策模块过于激进所有问题都触发搜索和GPT-4调用。2. 未对输入输出Token进行限制。3. 被恶意用户高频调用。1.精细化决策优化决策提示词加入更多否定案例进行训练降低误触发率。2.设置Token限制在调用OpenAI API时明确设置max_tokens参数防止生成长篇大论。3.实现速率限制在FastAPI层面使用中间件对IP或用户进行速率限制如slowapi。4.监控与告警记录每次调用的Token消耗和费用设置每日费用预算告警。5.2 高级优化方向当你解决了基本问题后可以考虑以下方向进一步提升系统能力混合检索策略不仅仅依赖必应搜索。可以结合本地知识库对于高频、稳定的内部知识如产品手册、公司制度优先从本地向量数据库检索速度更快、成本为零。多搜索引擎同时调用多个搜索API如Google Search API、SerpAPI综合结果提高信息覆盖面和可靠性。搜索结果重排序Re-ranking必应返回的结果顺序可能不是最相关的。可以引入一个轻量级的重排序模型如BAAI/bge-reranker根据用户问题对搜索结果进行相关性重排将最相关的结果放在前面提升上下文质量。流式输出Streaming对于生成时间较长的答案可以实现Server-Sent Events (SSE) 流式输出让用户看到答案逐字生成的过程极大提升体验。对话历史管理将当前项目改造成支持多轮对话。需要维护一个会话历史并在每次决策和搜索时考虑之前的对话上下文避免重复搜索或回答矛盾。可观测性与评估建立监控面板跟踪关键指标平均响应时间、搜索触发率、API调用成功率、用户反馈如有。定期用一批标准问题测试系统评估答案准确率驱动持续迭代。5.3 我的实操心得与避坑总结提示词需要“数据驱动”迭代不要指望一次写出完美的提示词。准备一个包含几十个问题的测试集手动标注预期行为和答案。每次修改提示词后跑一遍测试集用客观指标如搜索决策准确率、答案满意度来衡量改进。这是一个持续的优化过程。错误处理要面向用户网络错误、API限额、模型生成内容违规等错误后端要妥善捕获但返回给前端或用户的信息应该是友好的。例如不要直接抛出一段Python异常栈而是返回“服务暂时繁忙请稍后再试”或“您的问题可能涉及复杂信息当前无法处理”。关注Token消耗与成本尤其是在使用GPT-4时Token就是钱。在开发阶段多用gpt-3.5-turbo进行测试。上线后对输入用户问题搜索上下文的长度要保持敏感。过长的上下文不仅贵还可能影响模型性能。务必对输入文本进行必要的截断或总结。安全与内容过滤这是一个开放系统用户可能输入任何内容。务必在将用户问题传递给搜索引擎和GPT之前进行一层基本的内容安全过滤防止滥用。同时也要对GPT生成的内容进行审查避免输出有害信息。从“能用”到“好用”初期聚焦核心链路跑通。之后优化点会转移到体验上响应速度、答案的流畅度和准确性、引用格式的美观度、对用户追问的理解等。这些细节决定了产品的口碑。通过以上从架构到代码从原理到实操的详细拆解你应该对如何构建一个类似“bujnlc8/gptbing”的联网搜索AI助手有了全面的认识。这个项目是一个非常好的起点你可以在此基础上根据具体的应用场景融入更多想法和技术打造出更强大、更智能的AI应用。