feat: 添加深度分析智能体模块及相关上下文和工具

- 在 `src/agents/deep_agent` 中新增深度分析智能体模块,包含任务规划、深度知识搜索和分析能力。
- 引入 `deepagents` 库,创建深度智能体并实现复杂任务处理。
- 更新 `pyproject.toml` 以添加 `deepagents` 依赖。
- 修改 `context.py` 和 `graph.py`,为深度智能体提供专用上下文和工具支持。
- 更新文档以反映新功能和模块的集成。
This commit is contained in:
Wenjie Zhang 2025-11-12 01:21:38 +08:00
parent cab57e7f4d
commit 5adff04865
8 changed files with 291 additions and 9 deletions

View File

@ -11,6 +11,8 @@
- 集成智能体评估,首先使用命令行来实现,然后考虑放在 UI 里面展示
- 开发与生产环境隔离,构建生产镜像 <Badge type="info" text="0.4" />
- 集成 LangFuse (观望) 添加用户日志与用户反馈模块,可以在 AgentView 中查看信息
- 集成 neo4j mcp (或者自己构建工具)
- 工具组件重构以支持 todofiles 等渲染。
### Bugs
- 部分异常状态下,智能体的模型名称出现重叠[#279](https://github.com/xerrors/Yuxi-Know/issues/279)

View File

@ -60,6 +60,7 @@ dependencies = [
"tomli-w",
"aiofiles>=24.1.0",
"aiohttp>=3.9.0",
"deepagents>=0.2.5",
]
[tool.ruff]
line-length = 120 # 代码最大行宽

View File

@ -6,8 +6,8 @@ import yaml
from fastapi import APIRouter, Body, Depends, HTTPException
from src.storage.db.models import User
from server.utils.auth_middleware import get_admin_user, get_superadmin_user
from src import config, graph_base
from server.utils.auth_middleware import get_admin_user
from src import config
from src.models.chat import test_chat_model_status, test_all_chat_models_status
from src.utils.logging_config import logger

View File

@ -31,12 +31,12 @@ class BaseContext:
thread_id: str = field(
default_factory=lambda: str(uuid.uuid4()),
metadata={"name": "线程ID", "configurable": False, "description": "用来描述智能体的角色和行为"},
metadata={"name": "线程ID", "configurable": False, "description": "用来唯一标识一个对话线程"},
)
user_id: str = field(
default_factory=lambda: str(uuid.uuid4()),
metadata={"name": "用户ID", "configurable": False, "description": "用来描述智能体的角色和行为"},
metadata={"name": "用户ID", "configurable": False, "description": "用来唯一标识一个用户"},
)
system_prompt: str = field(
@ -45,7 +45,7 @@ class BaseContext:
)
model: Annotated[str, {"__template_metadata__": {"kind": "llm"}}] = field(
default="siliconflow/Qwen/Qwen3-235B-A22B-Instruct-2507",
default=sys_config.default_model,
metadata={"name": "智能体模型", "options": [], "description": "智能体的驱动模型"},
)

View File

@ -12,6 +12,9 @@ from src import config, graph_base, knowledge_base
from src.utils import logger
search = TavilySearch(max_results=10)
search.metadata = {"name": "Tavily 网页搜索"}
@tool(name_or_callable="计算器", description="可以对给定的2个数字选择进行 add, subtract, multiply, divide 运算")
def calculator(a: float, b: float, operation: str) -> float:
try:
@ -69,9 +72,15 @@ def get_approved_user_goal(
return result
@tool(name_or_callable="查询知识图谱", description="使用这个工具可以查询知识图谱中包含的三元组信息。")
KG_QUERY_DESCRIPTION = """
使用这个工具可以查询知识图谱中包含的三元组信息
关键词query使用可能帮助回答这个问题的关键词进行查询不要直接使用用户的原始输入去查询
"""
@tool(name_or_callable="查询知识图谱", description=KG_QUERY_DESCRIPTION)
def query_knowledge_graph(query: Annotated[str, "The keyword to query knowledge graph."]) -> Any:
"""Use this to query knowledge graph, which include some food domain knowledge."""
"""使用这个工具可以查询知识图谱中包含的三元组信息。关键词query使用可能帮助回答这个问题的关键词进行查询不要直接使用用户的原始输入去查询。"""
try:
logger.debug(f"Querying knowledge graph with: {query}")
result = graph_base.query_node(query, hops=2, return_format="triples")
@ -91,8 +100,6 @@ def get_static_tools() -> list:
# 检查是否启用网页搜索
if config.enable_web_search:
search = TavilySearch(max_results=10)
search.metadata = {"name": "Tavily 网页搜索"}
static_tools.append(search)
return static_tools

View File

@ -0,0 +1,22 @@
"""Deep Agent - 深度分析智能体模块
基于 deepagents 库构建的深度分析智能体具备以下特性
- 任务规划和分解能力
- 深度知识搜索和分析
- 子智能体协作
- 文件系统和长期记忆
- 综合分析和报告生成
"""
from .context import DeepContext
from .graph import DeepAgent
__all__ = [
"DeepAgent",
"DeepContext",
]
# 模块元数据
__version__ = "1.0.0"
__author__ = "Yuxi-Know Team"
__description__ = "基于 create_deep_agent 的深度分析智能体"

View File

@ -0,0 +1,36 @@
"""Deep Agent Context - 基于BaseContext的深度分析上下文配置"""
from dataclasses import field
from src.agents.common.context import BaseContext
DEEP_PROMPT = """你是一个能够处理复杂、多步骤任务的深度分析代理。
你拥有以下资源
- 用于分解复杂任务的规划工具
- 用于存储上下文和长期记忆的文件系统
- 用于委派专业工作的子代理
- 知识图谱查询与分析工具
你的工作方式
1. 仔细分析用户的请求
2. 如果任务复杂则制定计划
3. 使用合适的工具和子代理
4. 提供全面且有充分推理的回应
5. 存储重要信息以备将来参考
在分析中注重深度准确性和全面性"""
class DeepContext(BaseContext):
"""
Deep Agent 的上下文配置继承自 BaseContext
专门用于深度分析任务的配置管理
"""
# 深度分析专用的系统提示词
system_prompt: str = field(
default=DEEP_PROMPT,
metadata={"name": "系统提示词", "description": "Deep智能体的角色和行为指导"},
)

View File

@ -0,0 +1,214 @@
"""Deep Agent - 基于create_deep_agent的深度分析智能体"""
import os
from typing import Literal
from deepagents import create_deep_agent
from src.agents.common import BaseAgent, load_chat_model
from src.agents.common.tools import search
from src.agents.common.middlewares import (
context_aware_prompt,
context_based_model,
inject_attachment_context,
)
from .context import DeepContext
from tavily import TavilyClient
# 最佳实践是初始化客户端一次并复用它。
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
# 用于进行研究的搜索工具
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["general", "news", "finance"] = "general",
include_raw_content: bool = False,
):
"""运行网络搜索"""
search_docs = tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
return search_docs
sub_research_prompt = """你是一位专注的研究员。你的工作是根据用户的问题进行研究。
进行彻底的研究然后用详细的答案回复用户的问题
只有你的最终答案会被传递给用户除了你的最终信息他们不会知道任何其他事情所以你的最终报告应该就是你的最终信息"""
research_sub_agent = {
"name": "research-agent",
"description": (
"用于研究更深入的问题。一次只给这个研究员一个主题。不要向这个研究员传递多个子问题。"
"相反,你应该将一个大主题分解成必要的组成部分,然后并行调用多个研究代理,每个子问题一个。"
),
"system_prompt": sub_research_prompt,
"tools": [internet_search],
}
sub_critique_prompt = """你是一位专注的编辑。你的任务是评论一份报告。
你可以在 `final_report.md` 找到这份报告
你可以在 `question.txt` 找到这份报告的问题/主题
用户可能会要求评论报告的特定方面请用详细的评论回复用户指出报告中可以改进的地方
如果有助于你评论报告你可以使用搜索工具来搜索信息
不要自己写入 `final_report.md`
需要检查的事项
- 检查每个部分的标题是否恰当
- 检查报告的写法是否像论文或教科书它应该是以文本为主不要只是一个项目符号列表
- 检查报告是否全面如果任何段落或部分过短或缺少重要细节请指出来
- 检查文章是否涵盖了行业的关键领域确保了整体理解并且没有遗漏重要部分
- 检查文章是否深入分析了原因影响和趋势提供了有价值的见解
- 检查文章是否紧扣研究主题并直接回答问题
- 检查文章是否结构清晰语言流畅易于理解
"""
critique_sub_agent = {
"name": "critique-agent",
"description": "用于评论最终报告。给这个代理一些关于你希望它如何评论报告的信息。",
"system_prompt": sub_critique_prompt,
}
# 用于引导代理成为专家级研究员的提示前缀
research_instructions = """你是一位专家级研究员。你的工作是进行彻底的研究,然后撰写一份精美的报告。
你应该做的第一件事是把原始的用户问题写入 `question.txt`以便你有一个记录
使用 research-agent 进行深入研究它会用详细的答案回应你的问题/主题
当你认为有足够的信息来撰写最终报告时就把它写入 `final_report.md`
你可以调用 critique-agent 来获取对最终报告的评论之后如果需要你可以做更多的研究并编辑 `final_report.md`
你可以根据需要重复这个过程直到你对结果满意为止
一次只编辑一个文件如果你并行调用这个工具可能会有冲突
以下是撰写最终报告的说明
<report_instructions>
关键确保答案的语言与人类信息的语言相同如果你制定了一个待办事项计划你应该在计划中注明报告应该使用什么语言
注意报告应该使用的语言是问题所在的语言而不是问题所涉及的国家/地区的语言
请根据整体研究简报创建一个详细的答案该答案应
1. 组织良好有恰当的标题# 用于标题,## 用于章节,### 用于子章节)
2. 包含研究中的具体事实和见解
3. 使用 [标题](URL) 格式引用相关来源
4. 提供平衡透彻的分析尽可能全面并包含与整体研究问题相关的所有信息使用你进行深入研究并期望得到详细全面的答案
5. 在末尾包含一个来源部分列出所有引用的链接
你可以用多种不同的方式来组织你的报告以下是一些例子
要回答一个要求你比较两件事物的问题你可以这样组织你的报告
1/ 引言
2/ 主题A概述
3/ 主题B概述
4/ A与B的比较
5/ 结论
要回答一个要求你返回一个事物列表的问题你可能只需要一个部分即整个列表
1/ 事物列表或表格
或者你可以选择将列表中的每一项都作为报告中的一个独立部分当被要求提供列表时你不需要引言或结论
1/ 项目1
2/ 项目2
3/ 项目3
要回答一个要求你总结一个主题给出一份报告或概述的问题你可以这样组织你的报告
1/ 主题概述
2/ 概念1
3/ 概念2
4/ 概念3
5/ 结论
如果你认为你可以用一个部分来回答问题你也可以这样做
1/ 答案
请记住章节是一个非常灵活和松散的概念你可以按照你认为最好的方式来组织你的报告包括上面没有列出的方式
确保你的各个部分是连贯的并且对读者来说是有意义的
对于报告的每个部分请执行以下操作
- 使用简单清晰的语言
- 对报告的每个部分使用 ## 作为章节标题Markdown 格式)
- 绝不要将自己称为报告的作者这应该是一份专业的报告不含任何自我指涉的语言
- 不要在报告中说你正在做什么只需撰写报告不要添加任何你自己的评论
- 每个部分的长度应足以用你收集到的信息预计各部分会长且详尽你正在撰写一份深入的研究报告用户会期望得到透彻的答案
- 在适当的时候使用项目符号来列出信息但默认情况下请以段落形式撰写
请记住
简报和研究可能是英文的但在撰写最终答案时你需要将这些信息翻译成正确的语言
确保最终答案报告的语言与消息历史中的人类信息语言相同
用清晰的 markdown 格式化报告结构合理并在适当的地方包含来源引用
<引用规则>
- 在你的文本中为每个唯一的 URL 分配一个引文编号
- ### 来源 结尾,列出每个来源及其对应的编号
- 重要提示无论你选择哪些来源最终列表中的来源编号都应连续无间断1,2,3,4...
- 每个来源都应该是列表中的一个独立行项目这样在 markdown 中它会被渲染成一个列表
- 示例格式
[1] 来源标题: URL
[2] 来源标题: URL
- 引用非常重要请确保包含这些内容并特别注意确保其正确性用户通常会使用这些引文来查找更多信息
</引用规则>
</report_instructions>
你可以使用一些工具
## `internet_search`
使用此工具对给定的查询进行网络搜索你可以指定结果数量主题以及是否包含原始内容
"""
class DeepAgent(BaseAgent):
name = "深度分析智能体"
description = "具备规划、深度分析和子智能体协作能力的智能体,可以处理复杂的多步骤任务"
context_schema = DeepContext
capabilities = [
"file_upload",
]
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.graph = None
self.checkpointer = None
def get_tools(self):
"""返回 Deep Agent 的专用工具"""
tools = [search]
return tools
async def get_graph(self, **kwargs):
"""构建 Deep Agent 的图"""
if self.graph:
return self.graph
# 获取上下文配置
context = self.context_schema.from_file(module_name=self.name)
# 使用 create_deep_agent 创建深度智能体
graph = create_deep_agent(
model=load_chat_model(context.model),
system_prompt=context.system_prompt,
tools=self.get_tools(),
subagents=[critique_sub_agent, research_sub_agent],
middleware=[
context_based_model, # 动态模型选择
context_aware_prompt, # 动态系统提示词
inject_attachment_context, # 附件上下文注入
],
checkpointer=await self._get_checkpointer(),
)
self.graph = graph
return graph