""" title: DeepSeek R1 SmartSearch author: byscott description: 在 OpenWebUI 添加 DeepSeek R1 的思维链、接入博查联网搜索（根据需要自动开启），并优化搜索输出质量 (仅支持0.5.6及以上版本) version: 0.1.0 licence: MIT references: - author: zgccrui title: DeepSeek R1 description: 在OpwenWebUI中显示DeepSeek R1模型的思维链。 - 仅支持0.5.6及以上版本 - author: atgehrhardt title: Live Search description: Implements live search functionality in OpenWebUI - author: liuboyangaa title: DeepSeek R1 with Search description: In OpenWebUI, displays the thought chain of the DeepSeek R1 model and searchs (version 0.5.6 or higher) - author: yantingbing title: DeepSeek R1 with Bocha search description: In OpenWebUI, displays the thought chain of the DeepSeek R1 model and Bocha searchs (version 0.5.6 or higher) """ import json import httpx import re from typing import AsyncGenerator, Callable, Awaitable from pydantic import BaseModel, Field import asyncio """#######################博查搜索START#######################""" from typing import List, Union, Generator, Iterator import requests from bs4 import BeautifulSoup import urllib.parse import copy """#######################博查搜索END#######################""" class Pipe: class Valves(BaseModel): DEEPSEEK_API_BASE_URL: str = Field( default="https://api.deepseek.com/v1", description="DeepSeek API的基础请求地址", ) DEEPSEEK_API_KEY: str = Field( default="", description="用于身份验证的DeepSeek API密钥，可从控制台获取" ) DEEPSEEK_API_MODEL: str = Field( default="deepseek-reasoner", description="API请求的模型名称，默认为 deepseek-reasoner ", ) AUX_API_BASE_URL: str = Field( default="https://api.deepseek.com/v1", description="辅助模型API基础请求地址", ) AUX_API_KEY: str = Field(default="", description="辅助模型API密钥") AUX_API_MODEL: str = Field( default="deepseek-chat", description="辅助模型名称，默认为 deepseek-chat ", ) Bocha_API_KEY: str = Field( default="", description="用于身份验证的Bocha API密钥，可从控制台获取" ) enable_bocha: bool = Field(default=True) bocha_results: int = Field(default=5) def __init__(self): self.valves = self.Valves() self.data_prefix = "data:" self.emitter = None self.type = "manifold" self.id = "engine_search" self.name = "engines/" self.search_result = "" def pipes(self): return [ { "id": self.valves.DEEPSEEK_API_MODEL, "name": self.valves.DEEPSEEK_API_MODEL, } ] async def pipe( self, body: dict, __event_emitter__: Callable[[dict], Awaitable[None]] = None, results=None, ) -> AsyncGenerator[str, None]: """主处理管道（已移除缓冲）""" thinking_state = {"thinking": -1} # 用于存储thinking状态 self.emitter = __event_emitter__ # 用于存储联网模式下返回的参考资料列表 stored_references = [] # 联网搜索供应商 0-无 1-火山引擎 2-PPLX引擎 search_providers = 0 waiting_for_reference = False # 验证配置 if not self.valves.DEEPSEEK_API_KEY: yield json.dumps({"error": "未配置API密钥"}, ensure_ascii=False) return # 准备请求参数 headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } try: # 模型ID提取 model_id = body["model"].split(".", 1)[-1] payload = {**body, "model": model_id} """#######################博查搜索START#######################""" user_input = self._extract_user_input(body) if not user_input: yield json.dumps( {"error": "No search query provided"}, ensure_ascii=False ) return if isinstance(results, str): try: results = int(results) except ValueError: yield json.dumps( {"error": "Invalid number of results '{results}'"}, ensure_ascii=False, ) return combined_input = copy.deepcopy(payload["messages"]) combined_input[-1][ "content" ] = f""" # 严格输出一位数字，0或1。 请根据用户输入判断你是否需要联网查询资料后才能输出，是则输出1，否则输出0。 注意： - 条例冲突情况下，优先联网； - 假定你的知识库都是落后信息； - 对于需要补充外部信息才能回答的问题，需要联网； - 知识类、文献类、法律类的问题，需要联网； - 天气类、新闻类、实时信息类等时效性问题，或提到[最近]，[今天]，[本周]，[这个月]，[几号]等时间信息，需要联网； - 如果用户问题为闲聊，或问题与内容源无关，或问题不需要额外信息帮助回答，无需联网； - 最后一个user role是用户最新输入的「用户问题」，其余均为「历史记录」 - 「历史记录」和「用户问题」无关联时，只通过「用户问题」判断，忽略「历史记录」； # 用户输入：{combined_input[-1]["content"]} """ enable_bocha = self._is_enable_search(combined_input) if self.valves.enable_bocha and enable_bocha: print("Calling Bocha search") show_result, self.search_result = self._search_bocha( user_input, results ) yield f""" <details> <summary>点击展开搜索结果</summary> {show_result} </details> """ # 获取用户输入 messages = body["messages"] user_input = messages[-1]["content"] # 合并搜索结果和用户输入 combined_input = f""" # 判断用户输入是否是前文对话的延续，如果是则判断搜索结果是否与前文对话有关，无关则请忽略所有搜索结果。 # 以下内容是基于用户发送的消息的搜索结果: {self.search_result} 在我给你的搜索结果中，每个结果都是[webpage X begin]...[webpage X end]格式的，X代表每篇文章的数字索引。 请在适当的情况下在句子末尾引用上下文。 请按照引用编号[citation:X]的格式在答案中对应部分引用上下文。 如果一句话源自多个上下文，请列出所有相关的引用编号，例如[citation:3][citation:5]，切记不要将引用集中在最后返回引用编号，而是在答案对应部分列出。 在回答时，请注意以下几点： - 如果搜索结果中没有对应信息则用自己的知识库补充。 - 并非搜索结果的所有内容都与用户的问题密切相关，你需要结合问题，对搜索结果进行甄别、筛选。 - 对于列举类的问题（如列举所有航班信息），尽量将答案控制在10个要点以内，并告诉用户可以查看搜索来源、获得完整信息。优先提供信息完整、最相关的列举项；如非必要，不要主动告诉用户搜索结果未提供的内容。 - 对于创作类的问题（如写论文），请务必在正文的段落中引用对应的参考编号，例如[citation:3][citation:5]，不能只在文章末尾引用。你需要解读并概括用户的题目要求，选择合适的格式，充分利用搜索结果并抽取重要信息，生成符合用户要求、极具思想深度、富有创造力与专业性的答案。你的创作篇幅需要尽可能延长，对于每一个要点的论述要推测用户的意图，给出尽可能多角度的回答要点，且务必信息量大、论述详尽。 - 如果回答很长，请尽量结构化、分段落总结。如果需要分点作答，尽量控制在5个点以内，并合并相关的内容。 - 对于客观类的问答，如果问题的答案非常简短，可以适当补充一到两句相关信息，以丰富内容。 - 你需要根据用户要求和回答内容选择合适、美观的回答格式，确保可读性强。 - 你的回答应该综合多个相关网页来回答，不能重复引用一个网页。 - 除非用户要求，否则你回答的语言需要和用户提问的语言保持一致。 # 用户消息为： {user_input} """ # 将合并后的输入替换掉原来的用户输入部分 payload["messages"][-1]["content"] = combined_input """#######################博查搜索END#######################""" # 处理消息以防止连续的相同角色 messages = payload["messages"] i = 0 while i < len(messages) - 1: if messages[i]["role"] == messages[i + 1]["role"]: # 插入具有替代角色的占位符消息 alternate_role = ( "assistant" if messages[i]["role"] == "user" else "user" ) messages.insert( i + 1, {"role": alternate_role, "content": "[Unfinished thinking]"}, ) i += 1 # 发起API请求 async with httpx.AsyncClient(http2=True) as client: async with client.stream( "POST", f"{self.valves.DEEPSEEK_API_BASE_URL}/chat/completions", json=payload, headers=headers, timeout=300, ) as response: # 错误处理 if response.status_code != 200: error = await response.aread() yield self._format_error(response.status_code, error) return # 流式处理响应 async for line in response.aiter_lines(): if not line.startswith(self.data_prefix): continue # 截取 JSON 字符串 json_str = line[len(self.data_prefix) :] # 去除首尾空格后检查是否为结束标记 if json_str.strip() == "[DONE]": return try: data = json.loads(json_str) except json.JSONDecodeError as e: error_detail = f"解析失败 - 内容：{json_str}，原因：{e}" yield self._format_error("JSONDecodeError", error_detail) return if search_providers == 0: # 处理参考资料 stored_references = data.get("references", []) + data.get( "citations", [] ) if stored_references: ref_count = len(stored_references) yield '<details type="search">\n' yield f"<summary>已搜索 {ref_count} 个网站</summary>\n" # 如果data中有references，则说明是火山引擎的返回结果 if data.get("references"): for idx, reference in enumerate(stored_references, 1): yield f'> {idx}. [{reference["title"]}]({reference["url"]})\n' yield "</details>\n" search_providers = 1 # 如果data中有citations，则说明是PPLX引擎的返回结果 elif data.get("citations"): for idx, reference in enumerate(stored_references, 1): yield f"> {idx}. {reference}\n" yield "</details>\n" search_providers = 2 choice = data.get("choices", [{}])[0] # 结束条件判断 if choice.get("finish_reason"): return # 状态机处理 state_output = await self._update_thinking_state( choice.get("delta", {}), thinking_state ) if state_output: yield state_output if state_output == "<think>": yield "\n" # 处理并立即发送内容 content = self._process_content(choice["delta"]) if content: # 处理思考状态标记 if content.startswith("<think>"): content = re.sub(r"^<think>", "", content) yield "<think>" await asyncio.sleep(0.1) yield "\n" elif content.startswith("</think>"): content = re.sub(r"^</think>", "", content) yield "</think>" await asyncio.sleep(0.1) yield "\n" # 处理参考资料 if search_providers == 1: # 火山引擎的参考资料处理 # 如果文本中包含"摘要"，设置等待标志 if "摘要" in content: waiting_for_reference = True yield content continue # 如果正在等待参考资料的数字 if waiting_for_reference: # 如果内容仅包含数字或"、" if re.match(r"^(\d+|、)$", content.strip()): numbers = re.findall(r"\d+", content) if numbers: num = numbers[0] ref_index = int(num) - 1 if 0 <= ref_index < len(stored_references): ref_url = stored_references[ref_index][ "url" ] else: ref_url = "" content = f"[[{num}]]({ref_url})" # 保持等待状态继续处理后续数字 # 如果遇到非数字且非"、"的内容且不含"摘要"，停止等待 elif not "摘要" in content: waiting_for_reference = False elif search_providers == 2: # PPLX引擎的参考资料处理 def replace_ref(m): idx = int(m.group(1)) - 1 if 0 <= idx < len(stored_references): return f"[[{m.group(1)}]]({stored_references[idx]})" return f"[[{m.group(1)}]]()" content = re.sub(r"\[(\d+)\]", replace_ref, content) yield content except Exception as e: yield self._format_exception(e) async def _update_thinking_state(self, delta: dict, thinking_state: dict) -> str: """更新思考状态机（简化版）""" state_output = "" if thinking_state["thinking"] == -1 and delta.get("reasoning_content"): thinking_state["thinking"] = 0 state_output = "<think>" elif ( thinking_state["thinking"] == 0 and not delta.get("reasoning_content") and delta.get("content") ): thinking_state["thinking"] = 1 state_output = "\n</think>\n\n" return state_output def _process_content(self, delta: dict) -> str: """直接返回处理后的内容""" return delta.get("reasoning_content", "") or delta.get("content", "") def _emit_status(self, description: str, done: bool = False) -> Awaitable[None]: """发送状态更新""" if self.emitter: return self.emitter( { "type": "status", "data": { "description": description, "done": done, }, } ) return None def _format_error(self, status_code: int, error: bytes) -> str: if isinstance(error, str): error_str = error else: error_str = error.decode(errors="ignore") try: err_msg = json.loads(error_str).get("message", error_str)[:200] except Exception: err_msg = error_str[:200] return json.dumps( {"error": f"HTTP {status_code}: {err_msg}"}, ensure_ascii=False ) def _format_exception(self, e: Exception) -> str: err_type = type(e).__name__ return json.dumps({"error": f"{err_type}: {str(e)}"}, ensure_ascii=False) """#######################博查搜索START#######################""" def _extract_user_input(self, body: dict) -> str: messages = body.get("messages", []) if messages: last_message = messages[-1] if isinstance(last_message.get("content"), list): for item in last_message["content"]: if item["type"] == "text": return item["text"] else: return last_message.get("content", "") return "" def _parse_response(self, response): if "data" in response: data = response["data"] if "webPages" in data: webPages = data["webPages"] if "value" in webPages: result = [ { "id": item.get("id", ""), "name": item.get("name", ""), "url": item.get("url", ""), "snippet": item.get("snippet", ""), "summary": item.get("summary", ""), "siteName": item.get("siteName", ""), "siteIcon": item.get("siteIcon", ""), "datePublished": item.get("datePublished", "") or item.get("dateLastCrawled", ""), } for item in webPages["value"] ] return result def _search_bocha(self, query: str, results=None) -> tuple[str, str]: if not self.valves.enable_bocha: return "Bocha search is disabled", None print("Searching with bocha") try: url = "https://api.bochaai.com/v1/web-search?utm_source=ollama" headers = { "Authorization": f"Bearer {self.valves.Bocha_API_KEY}", "Content-Type": "application/json", } count = self.valves.bocha_results payload = json.dumps( { "query": query, "summary": True, "freshness": "noLimit", "count": count, } ) response = requests.post(url, headers=headers, data=payload, timeout=5) response.raise_for_status() results_list = self._parse_response(response.json()) if not results_list: return f"No results found for: {query}", None show_results = "" formatted_results = "" # return results_list for i, result in enumerate(results_list[:count]): link = result["url"] title = result["name"] snippet = result["snippet"] summary = result["summary"] show_results += f"{i}. [{title}]({link}): {snippet}\n" formatted_results += f""" [webpage {i} begin] 标题：{title} 内容：{summary} URL: {link} [webpage {i} end] """ return show_results, formatted_results except Exception as e: return f"An error occurred while searching Bocha: {str(e)}", None """#######################博查搜索END#######################""" """#######################判断是否开启搜索START#######################""" def _is_enable_search(self, query: str) -> bool: try: url = f"{self.valves.AUX_API_BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {self.valves.AUX_API_KEY}", "Content-Type": "application/json", } payload = { "model": f"{self.valves.AUX_API_MODEL}", "stream": False, "messages": query, } response = requests.request("POST", url, json=payload, headers=headers) results = response.json() if "content" in response.text: return ( True if results["choices"][-1]["message"]["content"] == "1" else False ) else: return True except Exception as e: return f"报错: {str(e)}" """#######################判断是否开启搜索END#######################"""