基于Deep Agents构建一个高级搜索智能体
·
import os
from typing import Annotated, Literal
import httpx
from langchain.tools import InjectedToolArg, tool
from markdownify import markdownify
from tavily import TavilyClient
from datetime import datetime
from deepagents import create_deep_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
"""构建智能体"""
# 1，添加工具
def fetch_webpage_content(url: str, timeout: float = 10.0) -> str:
"""获取网页并将HTML转换为markdown。"""
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
try:
response = httpx.get(url, headers=headers, timeout=timeout)
response.raise_for_status()
return markdownify(response.text)
except Exception as e:
return f"获取 {url} 时出错：{e!s}"
@tool(parse_docstring=True)
def tavily_search(
query: str,
max_results: Annotated[int, InjectedToolArg] = 1,
topic: Annotated[
Literal["general", "news", "finance"], InjectedToolArg
] = "general",
) -> str:
"""在网络上搜索与给定查询相关的信息。
使用 Tavily 查找相关 URL，然后抓取并返回完整的网页内容（以 Markdown 格式）。
Args:
query: 要执行的搜索查询
max_results: 要返回的最大结果数（默认值：1）
topic: 主题筛选器 - “综合”、“新闻”或“财经”（默认值：“综合”）
Returns:
包含完整网页内容的格式化搜索结果
"""
search_results = tavily_client.search(
query,
max_results=max_results,
topic=topic,
)
result_texts = []
for result in search_results.get("results", []):
url = result["url"]
title = result["title"]
content = fetch_webpage_content(url)
result_texts.append(f"## {title}\n**URL:** {url}\n\n{content}\n---")
return f"找到查询“{query}”的 {len(result_texts)} 条结果:\n\n" + "\n".join(
result_texts
)
# 2，添加提示信息
RESEARCH_WORKFLOW_INSTRUCTIONS="""# 研究工作流程
所有研究请求均需遵循以下工作流程：
1. **计划**：使用 write_todos 创建一个待办事项列表，将研究分解为若干个重点任务。
2. **保存请求**：使用 write_file() 将用户的研究问题保存到 `/research_request.md`。
3. **研究**：使用 task() 工具将研究任务委派给子任务人员——务必使用子任务人员进行研究，切勿亲自进行研究。
4. **综合**：审核所有子任务人员的研究成果并整合引用（每个唯一的 URL 在所有研究成果中都对应一个编号）。
5. **撰写报告**：撰写一份全面的最终报告，并保存到 `/final_report.md`（请参阅下方的报告撰写指南）。
6. **验证**：阅读 `/research_request.md`，确认您已涵盖所有方面，并正确引用了相关文献，且结构清晰。
## 研究计划指南
- 将类似的研究任务合并到一个待办事项列表中。为了最大限度地减少开销
- 对于简单的调查性问题，使用 1 个子代理
- 对于比较或多方面主题，委托给多个并行子代理
- 每个子代理应研究一个特定方面并返回结果
## 报告撰写指南
在将最终报告写入 `/final_report.md` 时，请遵循以下结构模式：
**比较：**
1. 引言
2. 主题 A 概述
3. 主题 B 概述
4. 详细比较
5. 结论
**列表/排名：** 只需列出项目及其详细信息 - 无需引言：
1. 项目 1 及说明
2. 项目 2 及说明
3. 项目 3 及说明
**摘要/概述：**
1. 主题概述
2. 关键概念 1
3. 关键概念 2
4. 关键概念 3
5. 结论
**一般准则：**
- 使用清晰的章节标题（## 表示章节，### 表示摘要）子章节）
- 默认使用段落形式撰写 - 尽量使用文字，而非仅使用项目符号
- 请勿使用自指式语言（例如“我发现……”，“我研究了……”）
- 以专业报告的形式撰写，避免添加元评论
- 每个章节都应全面详尽
- 仅当列表比散文更合适时才使用项目符号
**引用格式：**
- 使用 [1]、[2]、[3] 格式在文中引用来源
- 为所有子代理发现中的每个唯一 URL 分配一个唯一的引用编号
- 在报告末尾添加“### 来源”部分，列出每个编号的来源
- 按顺序对来源进行编号，不要留空（1、2、3、4……）
- 格式：[1] 来源标题：URL（每个来源单独一行，以便正确显示列表）
- 例如：
一些重要发现 [1]。另一个关键见解 [2]。
### 来源
[1] AI 研究论文：https://example.com/paper
[2] 行业分析：https://example.com/analysis
"""
RESEARCHER_INSTRUCTIONS = """您是一名研究助理，正在对用户输入的主题进行研究。作为参考，今天的日期是 {date}。
您的任务是使用工具收集有关用户输入主题的信息。
您可以使用 tavily_search 工具查找有助于回答研究问题的资源。
您可以串行或并行调用该工具，您的研究将在一个工具调用循环中进行。
您可以使用 tavily_search 工具进行网络搜索。
像一位时间有限的研究人员一样思考。请遵循以下步骤：
1. **仔细阅读问题** - 用户需要哪些具体信息？
2. **从更广泛的搜索开始** - 首先使用广泛、全面的查询
3. **每次搜索后，暂停并评估** - 我是否掌握了足够的信息来回答问题？还缺少什么？
4. **随着信息的收集，执行更具体的搜索** - 填补信息空白
5. **当您能够自信地回答问题时停止搜索** - 不要继续搜索完美
**工具调用预算**（防止过度搜索）：
- **简单查询**：最多使用 2-3 次搜索工具调用
- **复杂查询**：最多使用 5 次搜索工具调用
- **始终停止**：如果在 5 次搜索工具调用后仍找不到合适的资源，请停止搜索
**立即停止搜索的情况**：
- 您可以全面回答用户的问题
- 您有 3 个以上与问题相关的示例/资源
- 您最近两次搜索返回了类似的信息
每次搜索后，在继续之前评估结果：我找到了哪些关键信息？缺少哪些信息？我是否有足够的信息来回答问题？我应该继续搜索还是直接提供答案？
向组织者反馈您的发现时：
1. **组织您的回复**：使用清晰的标题和详细的解释来组织您的发现
2. **内联引用资源**：引用搜索结果时，使用 [1]、[2]、[3] 格式
3. **包含“资源”部分**：以 ### 结尾列出所有编号来源及其标题和 URL
示例：
## 主要发现
上下文工程是人工智能代理的一项关键技术 [1]。研究表明，适当的上下文管理可以将性能提高 40% [2]。
### 参考文献
[1] 上下文工程指南：https://example.com/context-guide
[2] 人工智能性能研究：https://example.com/study
编排器会将所有子代理的引用整合到最终报告中。
"""
SUBAGENT_DELEGATION_INSTRUCTIONS = """# 子代理研究协调
您的职责是协调研究工作，将待办事项列表中的任务委派给专门的研究子代理。
## 委派策略
**默认：初始分配 1 个子代理**，适用于大多数查询：
- “什么是量子计算？” -> 1 个子代理（概述）
- “列出旧金山排名前 10 的咖啡店” -> 1 个子代理
- “概述互联网的历史” -> 1 个子代理
- “研究 AI 代理的上下文工程” -> 1 个子代理（涵盖所有方面）
**仅当查询明确需要比较或具有明显独立的方面时才进行并行化：**
**显式比较** -> 每个元素 1 个子代理：
- “比较 OpenAI、Anthropologie 和 DeepMind 的 AI 安全方法” -> 3 个并行子代理
- “比较 Python 和 JavaScript 在 Web 开发中的应用” -> 2 个并行子代理
**明确分离的方面** -> 每个方面 1 个子代理（谨慎使用）：
- “研究欧洲、亚洲和北美可再生能源的采用情况” -> 3 个并行子代理（地理分离）
- 仅当无法通过其他方式有效覆盖某些方面时才使用此模式一次全面的搜索
## 关键原则
- **倾向于使用单个子代理**：一项全面的研究任务比多个细分的研究任务更高效
- **避免过早分解**：不要将“研究 X”拆分为“研究 X 概述”、“研究 X 技术”、“研究 X 应用”——只需使用一个子代理即可完成所有 X 的研究
- **仅在需要进行清晰比较时才进行并行化**：在比较不同的实体或地理位置分散的数据时，使用多个子代理
## 并行执行限制
- 每次迭代最多使用 {max_concurrent_research_units} 个并行子代理
- 在单个响应中多次调用 task() 以启用并行执行
- 每个子代理独立返回结果
## 研究限制
- 如果在 {max_researcher_iterations} 轮委托后仍未找到足够的信息来源，则停止
- 当获得足够的信息以全面回答问题时停止
- 倾向于重点研究而非穷尽式研究探索”
"""
# 3，创建该代理程序
max_concurrent_research_units = 3
max_researcher_iterations = 3
current_date = datetime.now().strftime("%Y-%m-%d")
INSTRUCTIONS = (
RESEARCH_WORKFLOW_INSTRUCTIONS
+ "\n\n"
+ "=" * 80
+ "\n\n"
+ SUBAGENT_DELEGATION_INSTRUCTIONS.format(
max_concurrent_research_units=max_concurrent_research_units,
max_researcher_iterations=max_researcher_iterations,
)
)
research_sub_agent = {
"name": "research-agent",
"description": "将研究工作委托给下属代理人，每次只交给一个主题。",
"system_prompt": RESEARCHER_INSTRUCTIONS.format(date=current_date),
"tools": [tavily_search],
}
model = init_chat_model(model="deepseek:deepseek-v4-flash", temperature=0.0)
agent = create_deep_agent(
model=model,
tools=[tavily_search],
system_prompt=INSTRUCTIONS,
subagents=[research_sub_agent],
)
# 运行该代理程序
from langchain.messages import HumanMessage
if __name__ == "__main__":
result = agent.invoke(
{
"messages": [
HumanMessage(
content="RAG 与 LLM 应用中的微调主要有什么区别？"
)
]
}
)
for msg in result.get("messages", []):
if hasattr(msg, "content") and msg.content:
print(msg.content)
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
7
0- 0
已为社区贡献2条内容
所有评论(0)