kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(deep_agent): 重构深度分析智能体并优化MCP配置

Browse Source 
重构深度分析智能体的提示词和上下文处理逻辑，将提示词集中管理并优化结构
修改MCP服务器配置，使用本地npx命令替代远程HTTP服务
更新文档中的环境变量名称和配置说明
在Makefile中添加.env文件检查
This commit is contained in:
Wenjie Zhang 2025-12-02 16:47:21 +08:00
parent 11603efffe
commit 339484fe72
10 changed files with 220 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
AGENTS.md
View File

@ -46,5 +46,6 @@ docker compose exec api uv run python test/your_script.py # 放在 test 文件
**其他**：
- 使用 YUXI_SUPER_ADMIN_NAME / YUXI_SUPER_ADMIN_PASSWORD 调试接口
- 如果需要新建说明文档（仅开发者可见，非必要不创建），则保存在 `docs/vibe` 文件夹下面
- 代码更新后要检查文档部分是否有需要更新的地方，文档的目录定义在 `docs/.vitepress/config.mts` 中。文档应该更新最新版（`docs/latest`）

1
CLAUDE.md
View File

@ -46,5 +46,6 @@ docker compose exec api uv run python test/your_script.py # 放在 test 文件
**其他**：
- 使用 YUXI_SUPER_ADMIN_NAME / YUXI_SUPER_ADMIN_PASSWORD 调试接口
- 如果需要新建说明文档（仅开发者可见，非必要不创建），则保存在 `docs/vibe` 文件夹下面
- 代码更新后要检查文档部分是否有需要更新的地方，文档的目录定义在 `docs/.vitepress/config.mts` 中。文档应该更新最新版（`docs/latest`）

4
Makefile
View File

@ -4,6 +4,10 @@
PYTEST_ARGS ?=
start:
@if [ ! -f .env ]; then \
echo "Error: .env file not found. Please create it from .env.template"; \
exit 1; \
fi
docker compose up -d
stop:

2
docs/latest/advanced/agents-config.md
View File

@ -131,7 +131,7 @@ MCP_SERVERS = {
```python
MCP_SERVERS = {
"mcp_server_chart": {
"mcp-server-chart": {
"command": "npx",
"args": ["-y", "@antv/mcp-server-chart"],
"transport": "stdio"

2
src/agents/common/base.py
View File

@ -23,11 +23,11 @@ class BaseAgent:
name = "base_agent"
description = "base_agent"
capabilities: list[str] = [] # 智能体能力列表，如 ["file_upload", "web_search"] 等
context_schema: type[BaseContext] = BaseContext # 智能体上下文 schema
def __init__(self, **kwargs):
self.graph = None # will be covered by get_graph
self.checkpointer = None
self.context_schema = BaseContext
self.workdir = Path(sys_config.save_dir) / "agents" / self.module_name
self.workdir.mkdir(parents=True, exist_ok=True)
self._metadata_cache = None # Cache for metadata to avoid repeated file reads

16
src/agents/common/mcp.py
View File

@ -5,9 +5,8 @@ from collections.abc import Callable
from typing import Any, cast
import traceback
from langchain_mcp_adapters.client import ( # type: ignore[import-untyped]
MultiServerMCPClient,
)
from langchain_mcp_adapters.client import MultiServerMCPClient
from src.utils import logger
@ -21,7 +20,7 @@ MCP_SERVERS = {
"transport": "streamable_http",
},
# "zhipu-web-search-sse": {
# "url": f"https://open.bigmodel.cn/api/mcp/web_search/sse?Authorization={os.getenv('ZHIPU_API_KEY')}",
# "url": f"https://open.bigmodel.cn/api/mcp/web_search/sse?Authorization={os.getenv('ZHIPUAI_API_KEY')}",
# "transport": "streamable_http",
# },
# 这些 stdio 的 MCP server 需要在本地启动，启动的时候需要安装对应的包，需要时间
@ -42,10 +41,9 @@ async def get_mcp_client(
server_configs: dict[str, Any] | None = None,
) -> MultiServerMCPClient | None:
"""Initializes an MCP client with the given server configurations."""
configs = server_configs or MCP_SERVERS
try:
client = MultiServerMCPClient(configs) # pyright: ignore[reportArgumentType]
logger.info(f"Initialized MCP client with servers: {list(configs.keys())}")
client = MultiServerMCPClient(server_configs) # pyright: ignore[reportArgumentType]
logger.info(f"Initialized MCP client with servers: {list(server_configs.keys())}")
return client
except Exception as e:
logger.error("Failed to initialize MCP client: {}", e)
@ -63,7 +61,7 @@ async def get_mcp_tools(server_name: str, additional_servers: dict[str, dict] =
mcp_servers = MCP_SERVERS | (additional_servers or {})
try:
assert server_name in mcp_servers, f"Server {server_name} not found in MCP_SERVERS"
assert server_name in mcp_servers, f"Server {server_name} not found in ({list(mcp_servers.keys())})"
client = await get_mcp_client({server_name: mcp_servers[server_name]})
if client is None:
return []
@ -76,7 +74,7 @@ async def get_mcp_tools(server_name: str, additional_servers: dict[str, dict] =
logger.info(f"Loaded {len(tools)} tools from MCP server '{server_name}'")
return tools
except AssertionError as e:
logger.warning(f"Failed to load tools from MCP server '{server_name}': {e}")
logger.warning(f"[assert] Failed to load tools from MCP server '{server_name}': {e}")
return []
except Exception as e:
logger.error(f"Failed to load tools from MCP server '{server_name}': {e}, traceback: {traceback.format_exc()}")

97
src/agents/deep_agent/context.py
View File

@ -1,27 +1,96 @@
"""Deep Agent Context - 基于BaseContext的深度分析上下文配置"""
from dataclasses import field
from dataclasses import dataclass, field
from src.agents.common.context import BaseContext
DEEP_PROMPT = """你是一个能够处理复杂、多步骤任务的深度分析代理。
你拥有以下资源：
- 用于分解复杂任务的规划工具
- 用于存储上下文和长期记忆的文件系统
- 用于委派专业工作的子代理
- 知识图谱查询与分析工具
DEEP_PROMPT = """你是一位专家级研究员。你的工作是进行彻底的研究，然后撰写一份精美的报告。
你的工作方式：
1. 仔细分析用户的请求
2. 如果任务复杂，则制定计划
3. 使用合适的工具和子代理
4. 提供全面且有充分推理的回应
5. 存储重要信息以备将来参考
你应该做的第一件事是把原始的用户问题写入 `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>
你可以使用一些工具。
"""
@dataclass
class DeepContext(BaseContext):
"""
Deep Agent 的上下文配置，继承自 BaseContext

187
src/agents/deep_agent/graph.py
View File

@ -1,182 +1,62 @@
"""Deep Agent - 基于create_deep_agent的深度分析智能体"""
import os
from typing import Literal
from deepagents import create_deep_agent
from tavily import TavilyClient
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from src.agents.common import BaseAgent, load_chat_model
from src.agents.common.middlewares import (
context_aware_prompt,
context_based_model,
inject_attachment_context,
)
from src.agents.common.middlewares import context_based_model, inject_attachment_context
from src.agents.common.tools import search
from .context import DeepContext
# 最佳实践是初始化客户端一次并复用它。
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
from .prompts import DEEP_PROMPT
# 用于进行研究的搜索工具
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
search_tools = [search]
sub_research_prompt = """你是一位专注的研究员。你的工作是根据用户的问题进行研究。
进行彻底的研究，然后用详细的答案回复用户的问题
只有你的最终答案会被传递给用户。除了你的最终信息，他们不会知道任何其他事情，所以你的最终报告应该就是你的最终信息！"""
research_sub_agent = {
"name": "research-agent",
"description": (
"用于研究更深入的问题。一次只给这个研究员一个主题。不要向这个研究员传递多个子问题。"
"相反，你应该将一个大主题分解成必要的组成部分，然后并行调用多个研究代理，每个子问题一个。"
),
"system_prompt": sub_research_prompt,
"tools": [internet_search],
"system_prompt": (
"你是一位专注的研究员。你的工作是根据用户的问题进行研究。"
"进行彻底的研究，然后用详细的答案回复用户的问题，只有你的最终答案会被传递给用户。"
"除了你的最终信息，他们不会知道任何其他事情，所以你的最终报告应该就是你的最终信息！"
),
"tools": search_tools,
}
sub_critique_prompt = """你是一位专注的编辑。你的任务是评论一份报告。
你可以在 `final_report.md` 找到这份报告。
你可以在 `question.txt` 找到这份报告的问题/主题。
用户可能会要求评论报告的特定方面。请用详细的评论回复用户，指出报告中可以改进的地方。
如果有助于你评论报告，你可以使用搜索工具来搜索信息
不要自己写入 `final_report.md`。
需要检查的事项：
- 检查每个部分的标题是否恰当
- 检查报告的写法是否像论文或教科书——它应该是以文本为主，不要只是一个项目符号列表！
- 检查报告是否全面。如果任何段落或部分过短，或缺少重要细节，请指出来。
- 检查文章是否涵盖了行业的关键领域，确保了整体理解，并且没有遗漏重要部分。
- 检查文章是否深入分析了原因、影响和趋势，提供了有价值的见解
- 检查文章是否紧扣研究主题并直接回答问题
- 检查文章是否结构清晰、语言流畅、易于理解。
"""
critique_sub_agent = {
"name": "critique-agent",
"description": "用于评论最终报告。给这个代理一些关于你希望它如何评论报告的信息。",
"system_prompt": sub_critique_prompt,
"system_prompt": (
"你是一位专注的编辑。你的任务是评论一份报告。\n\n"
"你可以在 `final_report.md` 找到这份报告。\n\n"
"你可以在 `question.txt` 找到这份报告的问题/主题。\n\n"
"用户可能会要求评论报告的特定方面。请用详细的评论回复用户，指出报告中可以改进的地方。\n\n"
"如果有助于你评论报告，你可以使用搜索工具来搜索信息\n\n"
"不要自己写入 `final_report.md`。\n\n"
"需要检查的事项：\n"
"- 检查每个部分的标题是否恰当\n"
"- 检查报告的写法是否像论文或教科书——它应该是以文本为主，不要只是一个项目符号列表！\n"
"- 检查报告是否全面。如果任何段落或部分过短，或缺少重要细节，请指出来。\n"
"- 检查文章是否涵盖了行业的关键领域，确保了整体理解，并且没有遗漏重要部分。\n"
"- 检查文章是否深入分析了原因、影响和趋势，提供了有价值的见解\n"
"- 检查文章是否紧扣研究主题并直接回答问题\n"
"- 检查文章是否结构清晰、语言流畅、易于理解。"
),
}
# 用于引导代理成为专家级研究员的提示前缀
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`
使用此工具对给定的查询进行网络搜索。你可以指定结果数量、主题以及是否包含原始内容。
"""
@dynamic_prompt
def context_aware_prompt(request: ModelRequest) -> str:
"""从 runtime context 动态生成系统提示词"""
return DEEP_PROMPT + "\n\n\n" + request.runtime.context.system_prompt
class DeepAgent(BaseAgent):
name = "深度分析智能体"
description = "具备规划、深度分析和子智能体协作能力的智能体，可以处理复杂的多步骤任务"
context_schema = DeepContext
capabilities = [
"file_upload",
"todo",
@ -188,9 +68,9 @@ class DeepAgent(BaseAgent):
self.graph = None
self.checkpointer = None
def get_tools(self):
async def get_tools(self):
"""返回 Deep Agent 的专用工具"""
tools = [search]
tools = search_tools
return tools
async def get_graph(self, **kwargs):
@ -204,8 +84,7 @@ class DeepAgent(BaseAgent):
# 使用 create_deep_agent 创建深度智能体
graph = create_deep_agent(
model=load_chat_model(context.model),
system_prompt=context.system_prompt,
tools=self.get_tools(),
tools=await self.get_tools(),
subagents=[critique_sub_agent, research_sub_agent],
middleware=[
context_based_model, # 动态模型选择

85
src/agents/deep_agent/prompts.py Normal file
View File

@ -0,0 +1,85 @@
DEEP_PROMPT = """你是一位专家级研究员。你的工作是进行彻底的研究，然后撰写一份精美的报告。
你应该做的第一件事是把原始的用户问题写入 `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>
你可以使用一些工具。
"""

7
src/agents/reporter/graph.py
View File

@ -8,9 +8,10 @@ from src.utils import logger
_mcp_servers = {
"mcp-server-chart": {
"url": "https://mcp.api-inference.modelscope.net/9993ae42524c4c/mcp",
"transport": "streamable_http",
},
"command": "npx",
"args": ["-y", "@antv/mcp-server-chart"],
"transport": "stdio"
}
}