# -*- coding: utf-8 -*- """ 研究 """ from typing import AsyncGenerator from asyncpg import Pool import httpx from pydantic import BaseModel, Field, cast from pydantic_ai import Agent, ModelMessage, RunContext from pydantic_ai.usage import RunUsage from pydantic_ai.messages import AgentStreamEvent from pydantic_ai.run import AgentRunResultEvent from pydantic_ai.toolsets import FunctionToolset from sqlmodel import REAL from application.workshop.models import DEEPSEEK_V4_FLASH_RESPONSE_MODEL from application.workshop.database import connection_pool class Deps(BaseModel): """ 依赖类 """ model_config = {"arbitrary_types_allowed": True} # 允许任意类型 connection_pool: Pool = Field(..., description="连接池") class Args(BaseModel): query: str = Field(description="查询语句") # 实例化工具集 bocha_web_search = FunctionToolset() @bocha_web_search.tool(name="联网搜索") # type: ignore async def web_search(ctx: RunContext[Deps], args: Args) -> str: """ 联网搜索工具: 用于查询互联网实时信息、新闻、政策法规、行业报告、企业公告、最新数据。 适合查询模型静态知识库不存在的外部网页信息。 回答时必须标注引用来源URL,禁止编造链接。 不要用于简单常识、数学计算、代码编写等无需联网的任务。 """ try: async with httpx.AsyncClient(timeout=60) as client: response = await client.post( "https://api.bocha.cn/v1/web-search", headers={ "Authorization": f"Bearer sk-a9ed73d8b26745799cf970f84d70f512", "Content-Type": "application/json", }, json={ "query": args.query, "summary": True, # 显示文本摘要 }, ) response.raise_for_status() # 解析响应载荷 response_payload = response.json() except httpx.HTTPStatusError as exception: return f"请求错误 {exception.response.status_code}: {exception.response.text}" except Exception as exception: return f"请求异常: {str(exception)}" if (code := response_payload.get("code")) != 200: return f"业务错误 {code}: {response_payload.get("msg", "未知业务错误")}" web_pages = response_payload.get("data", {}).get("webPages", {}).get("value", []) if not web_pages: return "未搜索到相关网页" result = [] # 搜索结果 for idx, web_page in enumerate(web_pages, 1): # 解析网页的标题 name = web_page.get("name", "") # 解析网页的URL url = web_page.get("url", "") # 解析网页的网站名称 site_name = web_page.get("siteName", "") # 解析网页的发布时间 date_published = web_page.get("datePublished") or web_page.get( "dateLastCrawled", "" ) # 优先解析网页内容的文本摘要,若文本摘要为空则解析简短描述 summary = web_page.get("summary") or web_page.get("snippet", "") result.append( f"### 搜索结果 {idx}\n" f"- 标题:{name}\n" f"- URL:{url}\n" f"- 来源网站:{site_name}\n" f"- 发布时间:{date_published}\n" f"- 摘要:{summary}" ) return "\n\n".join(result) instruction = """ 你是研究助手,可调用联网搜索工具。 ## 一、联网判断 需要联网:时效类信息、新闻、声明公告、政策法规、最新数据,以及模型静态知识库无法可靠回答的问题。 无需联网:基础常识、数学计算、代码编写与解释、用户已提供完整材料的分析任务、无时间依赖的静态问题。 【搜索硬约束】 1. 最多**2轮联网搜索**,严禁发起第3轮;优先在第一轮构造高信息密度的Query尝试获取完整答案,非必要不执行第二轮搜索。 2. 发起新一轮搜索前,必须评估已获取的搜索摘要,现有信息足以回答问题则立即停止调用工具。 3. 禁止仅改写同义词、做语义几乎等价的重复搜索;若存在局部信息缺失,优先从已有结果中提取可用内容,不要直接发起新搜索。 ## 二、时间与Query规则 时效判断严格以运行时传入的当前日期、年份、时区为准,禁止依靠模型内置记忆猜测时间。 - 用户未指定年份:**禁止自行向搜索Query中追加年份**。 - 用户明确给出年份:搜索Query必须完整保留用户指定的年份。 ## 三、搜索效率 默认只执行1次真实联网搜索;拿到搜索结果后评估证据,满足以下任意条件,必须停止搜索: - 现有信息足以作答;官方一手资料已给出答案;多个可靠来源结论一致;继续搜索仅起到重复确认的作用。 ❌禁止以“再次确认、查找更多来源、补充更多资料”等理由发起额外搜索。 ✅仅在以下4种场景,才允许执行第2轮搜索: ①第一轮无有效搜索结果;②回答问题必需的关键事实完全缺失;③权威信源之间事实存在明显冲突;④第一轮搜索结果严重偏题。 第二轮搜索必须瞄准明确的信息缺口,禁止全盘复刻上一轮查询;**不允许单次回合同时生成多条搜索请求**。 严格执行流程:执行搜索 → 阅读解析结果 → 判断信息缺口 → 再决定是否发起二次搜索。 ## 四、信源优先级 1级(最高):官网、官方文档、官方公告、一手原始资料 2级:权威媒体、专业机构、科研机构 3级:其他可信赖网页 官方资料已经可以完整回答问题时,无需再搜索其他来源做重复验证。 自媒体、论坛、转载内容,不可作为重大事实的唯一证据。 ## 五、最新内容校验 查询“最新”类问题,不可仅凭网页排序判断时效性;必须核对资料的**发布时间、版本时间,结合运行时提供的当前日期综合判断**。 当存在多份结果时,优先选用发布时间更近、信源等级更高的资料。 ## 六、事实规范 - 【事实】:搜索返回资料原文直接支持的内容 - 【推论】:基于已确认事实做合理推导 - 【推测】:证据不足前提下的主观判断 严禁编造URL、日期、新闻事件、数据、官方声明等任何信息;证据不足时直接如实说明,不得强行得出结论。 ## 七、引用规范 引用编号区分搜索轮次:S1‑1、S1‑2 代表第1轮搜索结果;S2‑1、S2‑2 代表第2轮搜索结果。 只引用回答中实际用到的资料。 引用格式:`[参考 S1‑1] https://xxx`,链接必须来源于 **web_search** 工具返回结果,禁止编造链接。 ## 八、输出规则 简单事实类问题:优先给出结论,附带关键证据与来源,回答简洁,避免冗余。 复杂研究类问题,按「关键证据 → 分析 → 结论 → 参考来源」结构组织内容;先陈列证据再输出总结,结论不能超出现有资料可支撑的范围。 """ # 智能体 agent = Agent[Deps, str]( model=DEEPSEEK_V4_FLASH_RESPONSE_MODEL, deps_type=Deps, output_type=str, # 若输出类型为 str,则智能体输出结果时不进行输出类型校验。若定义输出类型,需考虑处理相应事件 instructions=instruction, toolsets=[bocha_web_search], ) async def run_stream_events( user_prompt: str | None, message_history: list[ModelMessage], usage: RunUsage, ) -> AsyncGenerator[AgentStreamEvent | AgentRunResultEvent, None]: """ 运行并流式输出事件 :param user_prompt: 用户提示词 :param message_history: 消息历史 :param usage: 使用量 :return: AsyncGenerator """ if connection_pool is None: return async with agent.run_stream_events( deps=Deps(connection_pool=connection_pool), user_prompt=user_prompt, message_history=message_history, usage=usage, ) as events: async for event in events: print(event) yield event