1

agent/application/application.py

@@ -10,6 +10,21 @@ from application.workshop.database import connection_pool_init
 
 app = rx.App(
     theme=rx.theme(color_mode="light"),
+    head_components=[
+        rx.script(
+            """
+            document.addEventListener('click', function(e) {
+                const link = e.target.closest('a');
+                if (!link) return;
+                const href = link.getAttribute('href');
+                if (href && (href.startsWith('http://') || href.startsWith('https://'))) {
+                    link.target = '_blank';
+                    link.rel = 'noopener noreferrer';
+                }
+            })
+            """
+        )
+    ],
     stylesheets=["/styles.css"],
     style={
         "box-shadow": "none",
@@ -37,7 +52,7 @@ app = rx.App(
             "color": "var(--prismui-color-3)",
             "opacity": 1,
         },  # 覆盖占位符样式
-    },  # 自定义主题颜色
+    },
 )  # 此处变量名需使用 app ，具体原因目前尚不清楚
 
 # 添加页面：首页

agent/application/pages/conversation.py

@@ -376,7 +376,7 @@ def thinking(item: MessageHistoryItem) -> rx.Component:
     :return: Component
     """
     return rx.vstack(
-        # 思考的标题栏
+        # 标题栏
         rx.box(
             rx.hstack(
                 rx.text(
@@ -435,6 +435,7 @@ def tool_call(item: MessageHistoryItem) -> rx.Component:
     """
     return rx.vstack(
         # 标题栏
+        rx.box(
             rx.hstack(
                 rx.icon(
                     "square-function",
@@ -446,13 +447,30 @@ def tool_call(item: MessageHistoryItem) -> rx.Component:
                     font_size="var(--prismui-font-size-1)",
                     class_name=rx.cond(item.is_running, "is_running", None),
                 ),
+                rx.icon(
+                    "chevron-right",
+                    width="14px",
+                    height="14px",
+                    transform=rx.cond(item.is_shown, "rotate(90deg)", "rotate(0deg)"),
+                    transition="transform 0.18s ease-in-out",
+                ),
                 align_items="center",
                 gap="4px",
                 margin_bottom="8px",
                 line_height="22px",
             ),
+            # 点击事件，展示 / 隐藏消息历史项
+            on_click=lambda: ConversationState.toggle_message_history_item_shown(
+                item.id
+            ),
+            cursor="pointer",
+        ),
+        rx.box(
+            rx.box(
+                rx.cond(
+                    item.is_shown,
                     rx.hstack(
-            # 分割线
+                        # 垂线
                         rx.box(
                             width="1px",
                             margin="12px 7px",
@@ -462,12 +480,13 @@ def tool_call(item: MessageHistoryItem) -> rx.Component:
                             # 内容
                             rx.markdown(
                                 item.content,
+                                allow_html=True,
                                 width="100%",
                                 margin="4px 0",
                                 line_height="1.6",
                                 font_size="var(--prismui-font-size-1)",
                             ),
-                # 审批
+                            # 若有审批
                             rx.cond(
                                 item.has_approval,
                                 rx.vstack(
@@ -502,7 +521,9 @@ def tool_call(item: MessageHistoryItem) -> rx.Component:
                                         rx.button(
                                             "提交",
                                             # 点击事件：审批
-                                on_click=lambda: ConversationState.run(True, item.id),
+                                            on_click=lambda: ConversationState.run(
+                                                True, item.id
+                                            ),
                                             padding="0 12px",
                                             background_color="var(--prismui-background-color-7)",
                                             border_radius="var(--prismui-border-radius-3)",
@@ -530,8 +551,17 @@ def tool_call(item: MessageHistoryItem) -> rx.Component:
                         margin_bottom="8px",
                         gap="4px",
                     ),
+                    rx.fragment(),
+                ),
+                min_height="0",
+            ),
+            display="grid",
+            grid_template_rows=rx.cond(item.is_shown, "1fr", "0fr"),
             width="100%",
-        margin_bottom="8px",
+            opacity=rx.cond(item.is_shown, "1", "0"),
+            transition="grid-template-rows 0.18s ease-in-out, opacity 0.18s ease-in-out",
+            overflow="hidden",
+        ),
     )
 
 

agent/application/preprocessing.py

@@ -56,7 +56,9 @@ feishu_server_client = (
 # 实例化词嵌器
 embedder = Embedder(
     SentenceTransformerEmbeddingModel(
-        str(Path(__file__).parent / "models" / "paraphrase-multilingual-MiniLM-L12-v2"),
+        (
+            Path(__file__).parent / "models" / "paraphrase-multilingual-MiniLM-L12-v2"
+        ).as_posix(),
         settings=SentenceTransformersEmbeddingSettings(
             sentence_transformers_device="cpu",
             sentence_transformers_normalize_embeddings=True,
@@ -399,5 +401,5 @@ def query(
 
 if __name__ == "__main__":
     # 同步文档
-    sync_documents()
+    # sync_documents()
-    # print(query("需求评审有哪些流程"))
+    print(query("需求评审有哪些流程"))

agent/application/states/conversation.py

@@ -388,7 +388,7 @@ class ConversationState(rx.State):
 
                 # 运行并流式输出事件
                 stream_events = run_stream_events(
-                    user_id=self.user_id,
+                    deps=conversation.workflow.deps,
                     user_prompt=user_prompt,
                     message_history=message_history,
                     usage=usage,
@@ -402,6 +402,7 @@ class ConversationState(rx.State):
 
                 # 运行并流式输出事件
                 stream_events = run_stream_events(
+                    deps=conversation.workflow.deps,
                     user_prompt=user_prompt,
                     message_history=message_history,
                     usage=usage,
@@ -593,6 +594,7 @@ class ConversationState(rx.State):
                             message.approval = Approval(
                                 tool_call_id=tool_call_id,
                             )
+                            message.is_shown = True
                             yield  # 通知前端更新渲染
                             messages.append(message)
 
@@ -714,14 +716,16 @@ class ConversationState(rx.State):
                 from application.workshop.bank_customer_service import Deps
 
                 # 初始化银行客户服务工作流
-                workflow = Workflow(type=type)
+                workflow = Workflow(type=type, deps=Deps(user_id=self.user_id))
                 # 设置用户提示词
                 self.set_user_prompt(f"帮我查询账户余额")
 
             # 研究
             case WorkflowType.RESEARCH:
+                from application.workshop.research import Deps
+
                 # 初始化研究工作流
-                workflow = Workflow(type=type)
+                workflow = Workflow(type=type, deps=Deps())
 
             # 其他
             case _:

agent/application/workshop/bank_customer_service.py

@@ -2,7 +2,7 @@
 """
 银行客户服务
 """
-from typing import AsyncGenerator
+from typing import Any, AsyncGenerator
 
 from pydantic import BaseModel, Field
 from pydantic_ai import (
@@ -134,7 +134,7 @@ PartEndEvent(index=0, part=TextPart(content='您好，刘弼仁先生。您账
 
 
 async def run_stream_events(
-    user_id: str,
+    deps: Any,
     user_prompt: str | None,
     message_history: list[ModelMessage],
     usage: RunUsage,
@@ -143,10 +143,19 @@ async def run_stream_events(
 ) -> AsyncGenerator[AgentStreamEvent | AgentRunResultEvent[str] | None,]:
     """
     运行并流式输出事件
+    :param user_prompt: 用户提示词
+    :param message_history: 消息历史记录
+    :param usage: 运行使用情况
+    :param usage_limits: 运行使用限制
+    :param deferred_tool_results: 延迟工具结果
+    :return: 异步生成器，生成智能体流式事件或运行结果事件
     """
+    if not isinstance(deps, Deps):
+        deps = Deps.model_validate(deps)
+
     async with agent.run_stream_events(
         user_prompt=user_prompt,
-        deps=Deps(user_id=user_id),
+        deps=deps,
         message_history=message_history,
         usage=usage,
         usage_limits=usage_limits,

agent/application/workshop/database.py

@@ -26,5 +26,5 @@ async def connection_pool_init() -> AsyncIterator[None]:
         server_settings={"statement_timeout": "5000"},
     )
     yield
-    if connection_pool:
+    if connection_pool is not None:
         await connection_pool.close()

agent/application/workshop/research.py

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