1

agent/application/pages/conversation.py

@@ -298,8 +298,10 @@ def use_guidance() -> rx.Component:
                 workflow_button(WorkflowType.BOOK_FLIGHT),
                 # 工作流按钮：银行客户服务
                 workflow_button(WorkflowType.BANK_CUSTOMER_SERVICE),
-                # 工作流按钮：研究
+                # 工作流按钮：联网检索
-                workflow_button(WorkflowType.RESEARCH),
+                workflow_button(WorkflowType.WEB_RETRIEVE),
+                # 工作流按钮：知识库检索
+                workflow_button(WorkflowType.KNOWLEDGE_BASE_RETRIEVE),
                 flex_wrap="wrap",
                 align_items="center",
                 gap="12px",

agent/application/states/conversation.py

@@ -248,7 +248,7 @@ class ConversationState(rx.State):
         if not user_prompt.strip():
             return
 
-        conversation.user_prompt = user_prompt.strip()
+        conversation.user_prompt = user_prompt
 
     @rx.var
     def is_user_prompt_sending_button_disabled(self) -> bool:
@@ -394,9 +394,23 @@ class ConversationState(rx.State):
                     usage=usage,
                 )
 
-            # 研究
+            # 联网检索
-            case WorkflowType.RESEARCH:
+            case WorkflowType.WEB_RETRIEVE:
-                from application.workshop.research import (
+                from application.workshop.web_retrieve import (
+                    run_stream_events,
+                )
+
+                # 运行并流式输出事件
+                stream_events = run_stream_events(
+                    deps=conversation.workflow.deps,
+                    user_prompt=user_prompt,
+                    message_history=message_history,
+                    usage=usage,
+                )
+
+            # 知识库检索
+            case WorkflowType.KNOWLEDGE_BASE_RETRIEVE:
+                from application.workshop.knowledge_base_retrieve import (
                     run_stream_events,
                 )
 
@@ -720,11 +734,18 @@ class ConversationState(rx.State):
                 # 设置用户提示词
                 self.set_user_prompt(f"帮我查询账户余额")
 
-            # 研究
+            # 联网检索
-            case WorkflowType.RESEARCH:
+            case WorkflowType.WEB_RETRIEVE:
-                from application.workshop.research import Deps
+                from application.workshop.web_retrieve import Deps
 
-                # 初始化研究工作流
+                # 初始化联网检索工作流
+                workflow = Workflow(type=type, deps=Deps())
+
+            # 知识库检索
+            case WorkflowType.KNOWLEDGE_BASE_RETRIEVE:
+                from application.workshop.knowledge_base_retrieve import Deps
+
+                # 初始化知识库检索工作流
                 workflow = Workflow(type=type, deps=Deps())
 
             # 其他

agent/application/states/models.py

@@ -61,7 +61,8 @@ class WorkflowType(StrEnum):
     TALK = "对话"
     BOOK_FLIGHT = "预定航班"
     BANK_CUSTOMER_SERVICE = "银行客户服务"
-    RESEARCH = "研究"
+    WEB_RETRIEVE = "联网检索"
+    KNOWLEDGE_BASE_RETRIEVE = "知识库检索"
 
 
 class Workflow(BaseModel):

agent/application/workshop/knowledge_base_retrieve.py

@@ -0,0 +1,296 @@
+# -*- coding: utf-8 -*-
+"""
+知识库检索
+"""
+from asyncio import sleep
+from typing import Any, AsyncGenerator
+
+from httpx import AsyncClient, HTTPStatusError
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent, ModelMessage
+from pydantic_ai.messages import AgentStreamEvent
+from pydantic_ai.run import AgentRunResultEvent
+from pydantic_ai.toolsets import FunctionToolset
+from pydantic_ai.usage import RunUsage
+
+from application.workshop.database import connection_pool
+from application.workshop.models import DEEPSEEK_V4_FLASH_RESPONSE_MODEL
+
+
+class Deps(BaseModel):
+    """
+    依赖类
+    """
+
+    ...
+
+
+class Args(BaseModel):
+    text: str = Field(..., description="查询文本")
+    keywords: list[str] = Field(default_factory=list, description="查询关键词列表")
+
+
+# 实例化工具集
+toolset = FunctionToolset()
+
+
+@toolset.tool_plain()
+async def knowledge_base_retrieve(args: Args) -> str:
+    """
+    知识库检索：
+    用于基于知识库检索相关信息。
+    """
+    if connection_pool is None:
+        return "数据库连接池未初始化"
+
+    # --- 召回，先根据查询文本获取嵌入向量再基于查询关键词和文本向量混合检索 --- #
+    request_count = 0  # 请求次数
+    response_payload: dict | None = None
+    while request_count < 3:
+        request_count += 1
+        try:
+            async with AsyncClient(timeout=60) as client:
+                response = await client.post(
+                    "https://api.siliconflow.cn/v1/embeddings",
+                    headers={
+                        "Authorization": "Bearer sk-xsnuwirjjphhfdbvznfdfjqlinfdlrnlxuhkbbqynfnbhiqz",
+                        "Content-Type": "application/json",
+                    },
+                    json={
+                        "model": "BAAI/bge-m3",
+                        "input": args.text,
+                    },
+                )
+            # 检查响应状态
+            response.raise_for_status()
+            # 解析响应载荷
+            response_payload = response.json()
+            break
+
+        except HTTPStatusError as exception:
+            # 若触发限速则等待若干秒后重试
+            if exception.response.status_code == 429:
+                await sleep(2**request_count)
+                continue
+
+            raise RuntimeError(
+                f"HTTPError {exception.response.status_code}: {exception.response.text}"
+            )
+
+        except Exception as exception:
+            raise RuntimeError(f"Exception: {str(exception)}")
+
+    if response_payload is None:
+        raise RuntimeError("response_payload is None")
+    text_vector = response_payload["data"][0]["embedding"]
+
+    # 混合检索
+    async with connection_pool.acquire() as connection:
+        results = await connection.fetch(
+            """
+            WITH bm25_retrieve_results AS (
+                SELECT
+                    id,
+                    document_id,
+                    index,
+                    text,
+                    return_text,
+                    text_fingerprint,
+                    ROW_NUMBER() OVER (ORDER BY paradedb.score(id) DESC) AS bm25_retrieve_rank
+                FROM document_chunks
+                WHERE text @@@ $1
+                ORDER BY paradedb.score(id) DESC
+                LIMIT 50
+            ),
+            vector_retrieve_results AS (
+                SELECT
+                    id,
+                    document_id,
+                    index,
+                    text,
+                    return_text,
+                    text_fingerprint,
+                    ROW_NUMBER() OVER (ORDER BY text_vector <-> $2::vector(1024)) AS vector_retrieve_rank
+                FROM document_chunks
+                ORDER BY text_vector <-> $2::vector(1024)
+                LIMIT 50
+            ),
+            merged_results AS (
+                SELECT
+                    COALESCE(bm25_retrieve_results.id, vector_retrieve_results.id) AS id,
+                    COALESCE(bm25_retrieve_results.document_id, vector_retrieve_results.document_id) AS document_id,
+                    COALESCE(bm25_retrieve_results.index, vector_retrieve_results.index) AS index,
+                    COALESCE(bm25_retrieve_results.text, vector_retrieve_results.text) AS text,
+                    COALESCE(bm25_retrieve_results.return_text, vector_retrieve_results.return_text) AS return_text,
+                    COALESCE(bm25_retrieve_results.text_fingerprint, vector_retrieve_results.text_fingerprint) AS text_fingerprint,
+                    COALESCE(1.0/(bm25_retrieve_rank + 60), 0) + COALESCE(1.0/(vector_retrieve_rank + 60), 0) AS rank_score
+                FROM bm25_retrieve_results
+                FULL OUTER JOIN vector_retrieve_results USING(id)
+            ),
+            deduplication_results AS (
+                SELECT
+                    *,
+                    ROW_NUMBER() OVER (PARTITION BY text_fingerprint ORDER BY rank_score DESC) AS duplicate_index
+                FROM merged_results
+            )
+            SELECT
+                id,
+                document_id,
+                index,
+                text,
+                return_text,
+                rank_score
+            FROM deduplication_results
+            WHERE duplicate_index = 1
+            ORDER BY rank_score DESC
+            LIMIT 50;
+            """,
+            " ".join(args.keywords) if args.keywords else args.text,
+            "[" + ",".join(str(x) for x in text_vector) + "]",
+        )
+    if not results:
+        return "未检索到相关结果"
+    recall_results: list[dict] = [dict(row) for row in results]  # 召回结果
+
+    # --- 重排 --- #
+    request_count = 0  # 请求次数
+    response_payload: dict | None = None
+    while request_count < 3:
+        request_count += 1
+        try:
+            async with AsyncClient(timeout=60) as client:
+                response = await client.post(
+                    "https://api.siliconflow.cn/v1/rerank",
+                    headers={
+                        "Authorization": "Bearer sk-xsnuwirjjphhfdbvznfdfjqlinfdlrnlxuhkbbqynfnbhiqz",
+                        "Content-Type": "application/json",
+                    },
+                    json={
+                        "model": "BAAI/bge-reranker-v2-m3",
+                        "query": args.text,
+                        "documents": [
+                            result["return_text"] for result in recall_results
+                        ],
+                        "return_documents": False,
+                        "top_n": 10,
+                    },
+                )
+                # 检查响应状态
+                response.raise_for_status()
+                # 解析响应载荷
+                response_payload = response.json()
+                break
+
+        except HTTPStatusError as exception:
+            # 若触发限速则等待若干秒后重试
+            if exception.response.status_code == 429:
+                await sleep(2**request_count)
+                continue
+
+            raise RuntimeError(
+                f"HTTPError {exception.response.status_code}: {exception.response.text}"
+            )
+
+        except Exception as exception:
+            await sleep(2**request_count)
+            continue
+
+    if response_payload is None:
+        reranked_results: list[dict] = recall_results[:10]
+    else:
+        reranked_results = [
+            recall_results[result["index"]]
+            for result in response_payload["results"]
+            if result["relevance_score"] >= 0.6
+        ]
+
+    if not reranked_results:
+        return "未检索到相关结果"
+
+    # --- 重构结果 --- #
+    refactored_results: list[str] = [f"检索内容：**{args.text}**，检索结果：\n"]
+    for result in reranked_results:
+        refactored_results.append(f"{result['return_text']}\n")
+    return "\n".join(refactored_results)
+
+
+instruction = """
+你是内部知识库问答助手，可调用 knowledge_base_retrieve 工具检索已收录的文档片段。
+
+## 一、检索判断
+需要调用 knowledge_base_retrieve：问题涉及项目内部资料、技术文档、产品规格、历史方案、代码库说明、内部流程等知识库已收录的主题。
+无需调用：纯闲聊、基础常识、数学计算、代码编写与解释、用户已在对话中提供完整材料的分析任务。
+【检索硬约束】
+1. 最多**2轮知识库检索**，严禁发起第3轮。一轮定义：单次思考决策发起1次检索；**单次思考步骤最多只能生成1次工具调用，禁止同一思考周期并行发起多次检索**。必须串行：执行1次检索，拿到结果评估之后，才允许在后续思考中决定是否发起下一轮。
+2. 发起新一轮检索前，必须评估已获取的文档片段，现有片段足以回答问题则立即停止调用工具。
+3. 禁止仅改写同义词、做语义几乎等价的重复检索；若存在局部信息缺失，优先从已有片段中提取可用内容，不要直接发起新检索。
+4. 默认只执行1次真实检索；拿到结果后评估证据，满足以下任意条件，必须停止检索：
+- 现有片段足以作答；多个片段交叉印证同一结论；继续检索仅起到重复确认的作用。
+❌禁止以"再查一下""补充更多文档""确认一下"等理由发起额外检索。
+✅仅在以下4种场景，才允许执行第2轮检索：
+①第一轮无有效检索结果；②回答问题必需的关键事实在已有片段中完全缺失；③第一轮返回的片段与用户问题明显偏题；④第一轮只召回了泛泛结果，缺少问题指向的具体实体/术语。
+第二轮检索必须更换关键词组瞄准明确的信息缺口，禁止原样重发上一轮查询。
+> 重要：第二轮检索执行完成后，无论信息是否补齐，都必须终止检索，不再发起任何 knowledge_base_retrieve 调用。
+
+## 二、调用工具前的改写规范
+调用 knowledge_base_retrieve 时，禁止直接把用户原始疑问句透传，必须先完成：
+1. **去疑问化**：把"支付模块怎么设计的？"改写成陈述句"支付模块设计"填入 text。
+2. **指代消解**：结合对话历史，把"那个功能""之前那个报错""他说的方案"替换成具体名词后填入 text。
+3. **关键词提炼**：从问题中提取 2–5 个核心名词/术语/专有名词填入 keywords，例如：
+   - 用户问"退款流程里签名校验为什么会失败"
+   - → text="退款流程签名校验失败"
+   - → keywords=["退款流程", "签名校验"]
+4. keywords 只保留名词、术语、产品名、技术栈名，禁止包含虚词、疑问词、动词短语。
+5. 若问题本身就是一个名词短语（如"支付模块文档"），text 与 keywords 可一致。
+
+## 三、事实规范
+- 【事实】：检索返回文档片段原文直接支持的内容。
+- 【推论】：基于已确认片段做的合理推导。
+- 【推测】：证据不足前提下的主观判断，必须显式标注。
+严禁编造文档片段中不存在的结论、字段、参数、接口名、日期、数据。
+> 知识库未收录相关内容时，直接告知用户"知识库中未找到相关资料"，不得用模型自身记忆强行作答或补全。
+> 若2轮检索结束关键信息仍缺失，直接说明现有文档无法回答该问题，不编造。
+
+## 四、引用规范
+回答中引用知识库片段时，必须标注来源，格式：`[文档 document_id 片段 chunk_index]`，编号必须对应检索结果中实际返回的 document_id 与 chunk_index，禁止编造编号。
+只引用回答中实际用到的片段；不要把整段召回文本原样堆砌给用户，要先总结再引用。
+
+## 五、输出规则
+简单事实类问题：优先给出结论，附带关键片段与来源标注，回答简洁。
+复杂问题（多文档交叉、方案对比、流程梳理）：按「关键证据 → 分析 → 结论 → 来源标注」结构组织；先陈列依据再输出总结，结论不能超出现有片段可支撑的范围。
+"""
+
+# 智能体
+agent = Agent[Deps, str](
+    model=DEEPSEEK_V4_FLASH_RESPONSE_MODEL,
+    deps_type=Deps,
+    output_type=str,  # 若输出类型为 str，则智能体输出结果时不进行输出类型校验。若定义输出类型，需考虑处理相应事件
+    instructions=instruction,
+    toolsets=[toolset],
+)
+
+
+async def run_stream_events(
+    deps: Any,
+    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 not isinstance(deps, Deps):
+        deps = Deps.model_validate(deps)
+
+    async with agent.run_stream_events(
+        deps=deps,
+        user_prompt=user_prompt,
+        message_history=message_history,
+        usage=usage,
+    ) as events:
+        async for event in events:
+            yield event

agent/application/workshop/research.py

@@ -1,254 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
-研究
-"""
-from datetime import datetime
-from html import escape
-from typing import Any, AsyncGenerator
-
-import httpx
-from pydantic import BaseModel, Field
-from pydantic_ai import Agent, ModelMessage
-from pydantic_ai.messages import AgentStreamEvent
-from pydantic_ai.run import AgentRunResultEvent
-from pydantic_ai.toolsets import FunctionToolset
-from pydantic_ai.usage import RunUsage
-
-from application.workshop.database import connection_pool
-from application.workshop.models import DEEPSEEK_V4_FLASH_RESPONSE_MODEL
-
-
-class Deps(BaseModel):
-    """
-    依赖类
-    """
-
-    ...
-
-
-class Args(BaseModel):
-    keyword: str = Field(description="查询语句")
-
-
-# 实例化工具集
-bocha_web_search = FunctionToolset()
-
-
-@bocha_web_search.tool_plain()
-async def web_search(args: Args) -> str:
-    """
-    联网搜索工具：
-    用于查询互联网实时信息、新闻、政策法规、行业报告、企业公告、最新数据。
-    """
-    if connection_pool is None:
-        return "数据库连接池未初始化"
-
-    # 关键词
-    keyword = args.keyword
-
-    # 根据关键词查询联网搜索缓存
-    async with connection_pool.acquire() as conn:
-        results = await conn.fetchval(
-            "SELECT results FROM web_search_caches WHERE keyword = $1",
-            keyword,
-        )
-    # 若命中缓存则返回结果，否则先召回再精排、重构搜索结果
-    if results is not None:
-        return results
-
-    else:
-        # --- 召回阶段 --- #
-        try:
-            async with httpx.AsyncClient(timeout=60) as client:
-                response = await client.post(
-                    "https://api.bocha.cn/v1/web-search",
-                    headers={
-                        "Authorization": "Bearer sk-a9ed73d8b26745799cf970f84d70f512",
-                        "Content-Type": "application/json",
-                    },
-                    json={
-                        "query": keyword,
-                        "summary": True,  # 显示文本摘要
-                        "count": 20,  # 返回结果数
-                    },
-                )
-            # 检查响应状态
-            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", "未知业务错误")}"
-
-        # 召回网页
-        webpages = response_payload["data"]["webPages"]["value"]
-        if not webpages:
-            return "未搜索到相关网页"
-
-        # --- 精排阶段 --- #
-        try:
-            async with httpx.AsyncClient(timeout=60) as client:
-                response = await client.post(
-                    "https://api.siliconflow.cn/v1/rerank",
-                    headers={
-                        "Authorization": "Bearer sk-xsnuwirjjphhfdbvznfdfjqlinfdlrnlxuhkbbqynfnbhiqz",
-                        "Content-Type": "application/json",
-                    },
-                    json={
-                        "model": "BAAI/bge-reranker-v2-m3",
-                        "query": keyword,
-                        "documents": [
-                            called_webpage["summary"] or called_webpage["snippet"]
-                            for called_webpage in webpages
-                        ],
-                        "return_documents": True,  # 返回文档内容
-                        "top_n": 10,  # 返回结果数
-                    },
-                )
-            # 检查响应状态
-            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)}"
-
-        # 精排后网页
-        reranked_webpages = [
-            (result["index"], result["document"]["text"])
-            for result in response_payload["results"]
-        ]
-
-        # --- 重构搜索结果 --- #
-        results = []
-        results.append(f"搜索内容：**{keyword}**，搜索结果：\n")
-        for index, summary in reranked_webpages:
-            # 根据重排后网页索引获取网页
-            webpage = webpages[index]
-            # 解析网页的标题
-            name = escape(webpage["name"])
-            # 解析网页的URL
-            url = escape(webpage["url"])
-            # 解析网页的网站名称
-            site_name = escape(webpage["siteName"])
-            # 解析网页的发布时间
-            date_published = escape(
-                datetime.fromisoformat(webpage["datePublished"]).date().isoformat()
-            )
-
-            results.append(
-                f"- <a href='{url}'>**{name}**</a>({site_name}・{date_published})：{escape(summary)}"
-            )
-
-        results = "\n".join(results)
-
-        # 新增联网搜索缓存
-        async with connection_pool.acquire() as conn:
-            await conn.execute(
-                "INSERT INTO web_search_caches (keyword, results) VALUES ($1, $2)",
-                keyword,
-                results,
-            )
-
-    return results
-
-
-instruction = """
-你是研究助手，可调用联网搜索工具。
-## 一、联网判断
-需要联网：时效类信息、新闻、声明公告、政策法规、最新数据，以及模型静态知识库无法可靠回答的问题。
-无需联网：基础常识、数学计算、代码编写与解释、用户已提供完整材料的分析任务、无时间依赖的静态问题。
-
-【搜索硬约束】
-1. 最多**2轮联网搜索**，严禁发起第3轮。一轮定义：单次思考决策发起1条搜索Query；**单次思考步骤，最多只能生成1条搜索query，禁止同一思考周期并行发起多条联网搜索调用**。必须串行执行：执行1次搜索，拿到结果评估之后，才允许在后续思考中决定是否发起下一轮搜索。不允许一步同时调用多次web_search。
-2. 发起新一轮搜索前，必须评估已获取的搜索摘要，现有信息足以回答问题则立即停止调用工具。
-3. 禁止仅改写同义词、做语义几乎等价的重复搜索；若存在局部信息缺失，优先从已有结果中提取可用内容，不要直接发起新搜索。
-4. 默认只执行1次真实联网搜索；拿到搜索结果后评估证据，满足以下任意条件，必须停止搜索：
-- 现有信息足以作答；官方一手资料已给出答案；多个可靠来源结论一致；继续搜索仅起到重复确认的作用。
-❌禁止以“再次确认、查找更多来源、补充更多资料”等理由发起额外搜索。
-✅仅在以下4种场景，才允许执行第2轮搜索：
-①第一轮无有效搜索结果；②回答问题必需的关键事实完全缺失；③权威信源之间事实存在明显冲突；④第一轮搜索结果严重偏题。
-第二轮搜索必须瞄准明确的信息缺口，禁止全盘复刻上一轮查询。
-> 重要：第二轮搜索执行完成后，无论信息是否补齐，都必须终止搜索，不再发起任何web_search调用。
-
-## 二、时间与Query规则
-时效判断严格以运行时传入的当前日期、年份、时区为准，禁止依靠模型内置记忆猜测时间。
-- 用户未指定年份：**禁止自行向搜索Query中追加年份**。
-- 用户明确给出年份：搜索Query必须完整保留用户指定的年份。
-- 若检索到网页发布时间晚于运行时当前日期，直接舍弃该结果，不采信。
-
-## 三、信源优先级
-1级（最高）：官网、官方文档、官方公告、一手原始资料
-2级：权威媒体、专业机构、科研机构
-3级：其他可信赖网页
-> 媒体转述不等于官方原文，若官方原文存在，优先采信官方原文，不采用媒体转述版本。
-官方资料已经可以完整回答问题时，无需再搜索其他来源做重复验证。
-自媒体、论坛、转载内容，不可作为重大事实的唯一证据。
-
-## 四、最新内容校验
-查询“最新”类问题，不可仅凭网页排序判断时效性；必须核对资料的**发布时间、版本时间，结合运行时提供的当前日期综合判断**。
-当存在多份结果时，优先选用发布时间更近、信源等级更高的资料。
-
-## 五、事实规范
-- 【事实】：搜索返回资料原文直接支持的内容
-- 【推论】：基于已确认事实做合理推导
-- 【推测】：证据不足前提下的主观判断
-严禁编造URL、日期、新闻事件、数据、官方声明等任何信息；证据不足时直接如实说明，不得强行得出结论。
-> 若2轮搜索结束，关键信息仍然缺失，直接告知用户：现有检索资料无法获取目标信息，不编造内容。
-
-## 六、引用规范
-引用编号区分搜索轮次：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(
-    deps: Any,
-    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 not isinstance(deps, Deps):
-        deps = Deps.model_validate(deps)
-
-    async with agent.run_stream_events(
-        deps=deps,
-        user_prompt=user_prompt,
-        message_history=message_history,
-        usage=usage,
-    ) as events:
-        async for event in events:
-            print(event)
-            yield event

agent/application/workshop/web_retrieve.py

@@ -0,0 +1,260 @@
+# -*- coding: utf-8 -*-
+"""
+联网检索
+"""
+from datetime import datetime
+from html import escape
+from typing import Any, AsyncGenerator
+
+from httpx import AsyncClient, HTTPStatusError
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent, ModelMessage
+from pydantic_ai.messages import AgentStreamEvent
+from pydantic_ai.run import AgentRunResultEvent
+from pydantic_ai.toolsets import FunctionToolset
+from pydantic_ai.usage import RunUsage
+
+from application.workshop.database import connection_pool
+from application.workshop.models import DEEPSEEK_V4_FLASH_RESPONSE_MODEL
+
+"""
+联网检索缓存表建表SQL
+CREATE TABLE IF NOT EXISTS web_retrieve_caches (
+    id      UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    query   TEXT NOT NULL,
+    results TEXT NOT NULL
+);
+CREATE INDEX idx_web_retrieve_caches_query ON web_retrieve_caches (query);
+"""
+
+
+class Deps(BaseModel):
+    """
+    依赖类
+    """
+
+    ...
+
+
+class Args(BaseModel):
+    query: str = Field(..., description="查询语句")
+
+
+# 实例化工具集
+toolset = FunctionToolset()
+
+
+@toolset.tool_plain()
+async def web_retrieve(args: Args) -> str:
+    """
+    联网检索：
+    用于基于互联网检索实时信息、新闻、政策法规、行业报告、企业公告、最新数据。
+    """
+    if connection_pool is None:
+        return "数据库连接池未初始化"
+
+    # 查询
+    query = args.query
+
+    # 查询联网检索缓存
+    async with connection_pool.acquire() as connection:
+        results = await connection.fetchval(
+            "SELECT results FROM web_retrieve_caches WHERE query = $1",
+            query,
+        )
+    # 若命中缓存则返回结果，否则先召回、重排再返回重构结果
+    if results is not None:
+        return results
+
+    # --- 召回阶段 --- #
+    try:
+        async with AsyncClient(timeout=60) as client:
+            response = await client.post(
+                "https://api.bocha.cn/v1/web-search",
+                headers={
+                    "Authorization": "Bearer sk-a9ed73d8b26745799cf970f84d70f512",
+                    "Content-Type": "application/json",
+                },
+                json={
+                    "query": query,
+                    "summary": True,  # 显示文本摘要
+                    "count": 20,  # 返回结果数
+                },
+            )
+        # 检查响应状态
+        response.raise_for_status()
+        # 解析响应载荷
+        response_payload = response.json()
+
+    except 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", "未知业务错误")}"
+
+    # 召回网页
+    webpages = response_payload["data"]["webPages"]["value"]
+    if not webpages:
+        return "未检索到相关结果"
+
+    # --- 重排阶段 --- #
+    try:
+        async with AsyncClient(timeout=60) as client:
+            response = await client.post(
+                "https://api.siliconflow.cn/v1/rerank",
+                headers={
+                    "Authorization": "Bearer sk-xsnuwirjjphhfdbvznfdfjqlinfdlrnlxuhkbbqynfnbhiqz",
+                    "Content-Type": "application/json",
+                },
+                json={
+                    "model": "BAAI/bge-reranker-v2-m3",
+                    "query": query,
+                    "documents": [
+                        called_webpage["summary"] or called_webpage["snippet"]
+                        for called_webpage in webpages
+                    ],
+                    "return_documents": True,  # 返回文档内容
+                    "top_n": 10,  # 返回结果数
+                },
+            )
+        # 检查响应状态
+        response.raise_for_status()
+        # 解析响应载荷
+        response_payload = response.json()
+
+    except HTTPStatusError as exception:
+        return f"请求失败 {exception.response.status_code}: {exception.response.text}"
+
+    except Exception as exception:
+        return f"请求失败: {str(exception)}"
+
+    # 重排后网页
+    reranked_webpages = [
+        (result["index"], result["document"]["text"])
+        for result in response_payload["results"]
+    ]
+
+    # --- 重构检索结果 --- #
+    results = []
+    results.append(f"检索内容：**{query}**，检索结果：\n")
+    for index, summary in reranked_webpages:
+        # 根据重排后网页索引获取网页
+        webpage = webpages[index]
+        # 解析网页的标题
+        name = escape(webpage["name"])
+        # 解析网页的URL
+        url = escape(webpage["url"])
+        # 解析网页的网站名称
+        site_name = escape(webpage["siteName"])
+        # 解析网页的发布时间
+        date_published = escape(
+            datetime.fromisoformat(webpage["datePublished"]).date().isoformat()
+        )
+
+        results.append(
+            f"- <a href='{url}'>**{name}**</a>({site_name}・{date_published})：{escape(summary)}"
+        )
+
+    results = "\n".join(results)
+
+    # 创建联网检索缓存
+    async with connection_pool.acquire() as connection:
+        await connection.execute(
+            "INSERT INTO web_retrieve_caches (query, results) VALUES ($1, $2)",
+            query,
+            results,
+        )
+
+    return results
+
+
+instruction = """
+你是研究助手，可调用联网检索工具。
+## 一、联网判断
+需要联网：时效类信息、新闻、声明公告、政策法规、最新数据，以及模型静态知识库无法可靠回答的问题。
+无需联网：基础常识、数学计算、代码编写与解释、用户已提供完整材料的分析任务、无时间依赖的静态问题。
+
+【检索硬约束】
+1. 最多**2轮联网检索**，严禁发起第3轮。一轮定义：单次思考决策发起1条检索Query；**单次思考步骤，最多只能生成1条检索query，禁止同一思考周期并行发起多条联网检索调用**。必须串行执行：执行1次检索，拿到结果评估之后，才允许在后续思考中决定是否发起下一轮检索。不允许一步同时调用多次web_search。
+2. 发起新一轮检索前，必须评估已获取的检索摘要，现有信息足以回答问题则立即停止调用工具。
+3. 禁止仅改写同义词、做语义几乎等价的重复检索；若存在局部信息缺失，优先从已有结果中提取可用内容，不要直接发起新检索。
+4. 默认只执行1次真实联网检索；拿到检索结果后评估证据，满足以下任意条件，必须停止检索：
+- 现有信息足以作答；官方一手资料已给出答案；多个可靠来源结论一致；继续检索仅起到重复确认的作用。
+❌禁止以“再次确认、查找更多来源、补充更多资料”等理由发起额外检索。
+✅仅在以下4种场景，才允许执行第2轮检索：
+①第一轮无有效检索结果；②回答问题必需的关键事实完全缺失；③权威信源之间事实存在明显冲突；④第一轮检索结果严重偏题。
+第二轮检索必须瞄准明确的信息缺口，禁止全盘复刻上一轮查询。
+> 重要：第二轮检索执行完成后，无论信息是否补齐，都必须终止检索，不再发起任何web_search调用。
+
+## 二、时间与Query规则
+时效判断严格以运行时传入的当前日期、年份、时区为准，禁止依靠模型内置记忆猜测时间。
+- 用户未指定年份：**禁止自行向检索Query中追加年份**。
+- 用户明确给出年份：检索Query必须完整保留用户指定的年份。
+- 若检索到网页发布时间晚于运行时当前日期，直接舍弃该结果，不采信。
+
+## 三、信源优先级
+1级（最高）：官网、官方文档、官方公告、一手原始资料
+2级：权威媒体、专业机构、科研机构
+3级：其他可信赖网页
+> 媒体转述不等于官方原文，若官方原文存在，优先采信官方原文，不采用媒体转述版本。
+官方资料已经可以完整回答问题时，无需再检索其他来源做重复验证。
+自媒体、论坛、转载内容，不可作为重大事实的唯一证据。
+
+## 四、最新内容校验
+查询“最新”类问题，不可仅凭网页排序判断时效性；必须核对资料的**发布时间、版本时间，结合运行时提供的当前日期综合判断**。
+当存在多份结果时，优先选用发布时间更近、信源等级更高的资料。
+
+## 五、事实规范
+- 【事实】：检索返回资料原文直接支持的内容
+- 【推论】：基于已确认事实做合理推导
+- 【推测】：证据不足前提下的主观判断
+严禁编造URL、日期、新闻事件、数据、官方声明等任何信息；证据不足时直接如实说明，不得强行得出结论。
+> 若2轮检索结束，关键信息仍然缺失，直接告知用户：现有检索资料无法获取目标信息，不编造内容。
+
+## 六、引用规范
+引用编号区分检索轮次：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=[toolset],
+)
+
+
+async def run_stream_events(
+    deps: Any,
+    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 not isinstance(deps, Deps):
+        deps = Deps.model_validate(deps)
+
+    async with agent.run_stream_events(
+        deps=deps,
+        user_prompt=user_prompt,
+        message_history=message_history,
+        usage=usage,
+    ) as events:
+        async for event in events:
+            yield event

短视频合成自动化/main.py

@@ -14,5 +14,5 @@ if __name__ == "__main__":
 
     # 批量创建草稿
     drafts_manager.batch_create(
-        draft_counts=900,
+        draft_counts=400,
     )