AI创想
标题:
deepsearch笔记
[打印本页]
作者:
AI小编
时间:
2026-7-7 07:20
标题:
deepsearch笔记
作者:后季暖
前面是一些py基础的笔记。
1.pydantic基础语法
pydantic的=,和python的=不一样,这里的=都是默认值是什么的意思!
pydantic的...是必填的意思。这里...表示此字段必填。
class Location(BaseModel):
longitude: float = Field(...,description="经度")
latitude: float = Field(...,description="纬度")
复制代码
2.pydantic的验证功能
from pydantic import field_validator
class WeatherInfo(BaseModel):
temperature: int
@field_validator('temperature',mode='before')
def parse_temperature(cls,v):
"""解析温度字符串:"16°C" -> 16"""
if isinstance(v,str):
v = v.replace('°C','').replace('℃','').strip()
return int(v)
return v
复制代码
注意,
验证器只在创建对象时执行,创建后修改属性不会触发验证:
# ✅ 创建时经过验证
product = Product(discount="80%") # discount = 0.8 (float)
# ❌ 直接赋值绕过验证
product.discount = "白嫖" # 不会报错!不会触发 parse_discount
print(product.discount) # "白嫖" (字符串!)
复制代码
这就是为什么 Pydantic 官方建议把模型对象当作
不可变对象
来用,或者开启验证:
# 如果需要修改后也验证,用 model_config
class Product(BaseModel):
model_config = {'validate_assignment': True} # 属性赋值时也验证
discount: float
@field_validator('discount', mode='before')
def parse_discount(cls, v):
if isinstance(v, str) and '%' in v:
return float(v.replace('%', '')) / 100
return v
# 现在直接赋值也会触发验证
product = Product(discount=0.9)
product.discount = "80%" # ✅ 自动转换,discount = 0.8
product.discount = "白嫖" # ❌ ValidationError
复制代码
3.items() 就是 Python 字典(dict)里的一个功能
最简单的例子:
# 一个字典
person = {
"name": "小明",
"age": 20,
"city": "北京"
}
# 用 items() 遍历
for key, value in person.items():
print("键:", key, " 值:", value)
复制代码
4.缓存命中
提示缓存
许多提供商提供提示缓存功能,以减少对相同令牌重复处理的延迟和成本。这些功能可以是
隐式
或
显式
:
隐式提示缓存:
如果请求命中缓存,提供商将自动传递成本节省。示例:OpenAI 和 Gemini(Gemini 2.5 及以上)。
显式缓存:
提供商允许您手动指示缓存点以获得更大控制或保证成本节省。示例:ChatOpenAI(通过 prompt_cache_key)、Anthropic 的 AnthropicPromptCachingMiddleware 和 cache_control 选项、AWS Bedrock、Gemini。
警告
提示缓存通常仅在超过最小输入令牌阈值时才会启用。请参阅提供商页面以了解详细信息。
缓存使用情况将反映在模型响应的使用元数据中。
(, 下载次数: 11)
上传
点击文件名下载附件
(, 下载次数: 12)
上传
点击文件名下载附件
(, 下载次数: 10)
上传
点击文件名下载附件
(, 下载次数: 15)
上传
点击文件名下载附件
(, 下载次数: 13)
上传
点击文件名下载附件
(, 下载次数: 14)
上传
点击文件名下载附件
显示缓存的优势:
(, 下载次数: 12)
上传
点击文件名下载附件
(, 下载次数: 13)
上传
点击文件名下载附件
(, 下载次数: 11)
上传
点击文件名下载附件
get_usage_metadata_callback可以直接收集你每个模型的token用量。就按照上面的用法就行。不然你只能手动收集:
(, 下载次数: 7)
上传
点击文件名下载附件
其实观察上面的callback都是写在一个config里面的,然后config直接在invoke的时候放进去:
(, 下载次数: 12)
上传
点击文件名下载附件
文档位置查看有哪些config:RunnableConfig | langchain_core | LangChain Reference
tool message的artifact解释,下面是示例:
from langchain.messages import ToolMessage
# 发送给模型
message_content = "It was the best of times, it was the worst of times."
# 下游可用的 artifact
artifact = {"document_id": "doc_123", "page": 0}
tool_message = ToolMessage(
content=message_content,
tool_call_id="call_123",
name="search_books",
artifact=artifact,
)
复制代码
其实这段代码的真实执行顺序:
(, 下载次数: 10)
上传
点击文件名下载附件
(, 下载次数: 8)
上传
点击文件名下载附件
也就是说,artifact其实是tool函数内部返回的一个值,捎带出来的。模型不会去看,是给我们用的。
content_blocks
(, 下载次数: 6)
上传
点击文件名下载附件
(, 下载次数: 8)
上传
点击文件名下载附件
(, 下载次数: 12)
上传
点击文件名下载附件
不同模态的content_block不一样,不只是text,还有:
(, 下载次数: 11)
上传
点击文件名下载附件
type除了上面的推理,还有下面的各种多模态文件类型:
(, 下载次数: 15)
上传
点击文件名下载附件
(, 下载次数: 11)
上传
点击文件名下载附件
还有很多就忽略了,甚至还有tool的:
(, 下载次数: 12)
上传
点击文件名下载附件
还有比较少见的:
(, 下载次数: 11)
上传
点击文件名下载附件
(, 下载次数: 12)
上传
点击文件名下载附件
AgentState
(, 下载次数: 8)
上传
点击文件名下载附件
所以比如下面这个中间件,用来裁剪消息的,用到的agentstate就是上面的:
文档地址:短期记忆 | LangChain 中文文档
(, 下载次数: 15)
上传
点击文件名下载附件
REMOVE_ALL_MESSAGES就是删除所有的消息
(, 下载次数: 16)
上传
点击文件名下载附件
@after_model用来删减消息更好,因为能
先利用完整上下文,再清理
。
下面是总结消息的内置中间件。它解决了一个核心矛盾:删消息会丢信息,不删会超 token 限制。这个中间件的做法是:
当消息太长时,用一个小模型生成摘要,用摘要替代旧消息
。
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
checkpointer = InMemorySaver()
agent = create_agent(
model="openai:gpt-4o",
tools=[],
middleware=[
SummarizationMiddleware(
model="openai:gpt-4o-mini",
max_tokens_before_summary=4000, # Trigger summarization at 4000 tokens
messages_to_keep=20, # Keep last 20 messages after summary
)
],
checkpointer=checkpointer,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)
final_response["messages"][-1].pretty_print()
"""
================================== Ai Message ==================================
Your name is Bob!
"""
复制代码
在工具中读取短期记忆 (Read short-term memory in a tool)
使用
ToolRuntime
参数在工具中访问短期记忆(状态)。
tool_runtime 参数对工具签名是
隐藏
的(因此模型看不到它),但工具可以通过它访问状态。
这里的短期记忆其实就是状态,下面代码是我自定义了一个状态CustomState。
然后runtime.state就能获取到这个状态。如果你不定义自定义的状态,默认也有一个的。
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime
class CustomState(AgentState):
user_id: str
@tool
def get_user_info(
runtime: ToolRuntime
) -> str:
"""Look up user info."""
user_id = runtime.state["user_id"]
return "User is John Smith" if user_id == "user_123" else "Unknown user"
agent = create_agent(
model="openai:gpt-5-nano",
tools=[get_user_info],
state_schema=CustomState,
)
result = agent.invoke({
"messages": "look up user information",
"user_id": "user_123"
})
print(result["messages"][-1].content)
# > User is John Smith.
复制代码
下面两个是完全不一样的。有的时候你自定义state,有时候context,有时候一起都要自定义。
(, 下载次数: 14)
上传
点击文件名下载附件
(, 下载次数: 11)
上传
点击文件名下载附件
(, 下载次数: 17)
上传
点击文件名下载附件
流式传输
from langchain.agents import create_agent
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
return f"It's always sunny in {city}!"
agent = create_agent(
model="openai:gpt-5-nano",
tools=[get_weather],
)
for chunk in agent.stream( # [!code highlight]
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode="updates",
):
for step, data in chunk.items():
print(f"step: {step}")
print(f"content: {data['messages'][-1].content_blocks}")
复制代码
(, 下载次数: 8)
上传
点击文件名下载附件
chunk的结构:
# chunk 的内容示例:
{
"model": { # 步骤名(字符串键)
"messages": [ # 该步骤的更新数据(字典值)
AIMessage(
content_blocks=[...] # 模型生成的工具调用或文本
)
]
}
}
# 或者
{
"tools": { # 步骤名
"messages": [ # 更新数据
ToolMessage(
content="It's always sunny in SF!",
tool_call_id="call_123"
)
]
}
}
复制代码
这段代码展示了如何在工具执行过程中
自定义流式输出
,让你能实时推送工具内部的进度信息。
核心机制:get_stream_writer()
from langchain.agents import create_agent
from langgraph.config import get_stream_writer # [!code highlight]
def get_weather(city: str) -> str:
"""获取给定城市的天气。"""
writer = get_stream_writer() # [!code highlight]
# 流式传输任何任意数据
writer(f"Looking up data for city: {city}")
writer(f"Acquired data for city: {city}")
return f"It's always sunny in {city}!"
agent = create_agent(
model="anthropic:claude-sonnet-4-5",
tools=[get_weather],
)
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
stream_mode="custom" # [!code highlight]
):
print(chunk)
复制代码
(, 下载次数: 14)
上传
点击文件名下载附件
中间件笔记略,用到的时候可以参考:中间件 | LangChain 中文文档
provider strategy和tool strategy区别
(, 下载次数: 9)
上传
点击文件名下载附件
(, 下载次数: 12)
上传
点击文件名下载附件
(, 下载次数: 16)
上传
点击文件名下载附件
(, 下载次数: 14)
上传
点击文件名下载附件
(, 下载次数: 6)
上传
点击文件名下载附件
(, 下载次数: 10)
上传
点击文件名下载附件
(, 下载次数: 8)
上传
点击文件名下载附件
Tool Strategy异常处理
(, 下载次数: 11)
上传
点击文件名下载附件
守卫
人工审核,执行工具前会先问用户,得到批准才执行:
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
agent = create_agent(
model="openai:gpt-4o",
tools=[search_tool, send_email_tool, delete_database_tool],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
# 要求批准敏感操作
"send_email": True,
"delete_database": True,
# 自动批准安全操作
"search": False,
}
),
],
# 在中断期间持久化状态
checkpointer=InMemorySaver(),
)
# 人工审核需要一个线程 ID 来进行持久化
config = {"configurable": {"thread_id": "some_id"}}
# 在执行敏感工具之前,智能体将暂停并等待批准
result = agent.invoke(
{"messages": [{"role": "user", "content": "Send an email to the team"}]},
config=config
)
result = agent.invoke(
Command(resume={"decisions": [{"type": "approve"}]}),
config=config # 相同的线程 ID 以恢复暂停的对话
)
复制代码
(, 下载次数: 9)
上传
点击文件名下载附件
request、runtime、context等的关系:
(, 下载次数: 9)
上传
点击文件名下载附件
state只限于当前会话,但是store它可以跨会话的。所以store用来当长期记忆。
(, 下载次数: 14)
上传
点击文件名下载附件
context常见用法的示例:
根据用户的管辖范围,从运行时上下文中注入合规规则:
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
@dataclass
class Context:
user_jurisdiction: str
industry: str
compliance_frameworks: list[str]
@wrap_model_call
def inject_compliance_rules(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Inject compliance constraints from Runtime Context."""
# 从 Runtime Context 读取:获取合规性要求
jurisdiction = request.runtime.context.user_jurisdiction
industry = request.runtime.context.industry
frameworks = request.runtime.context.compliance_frameworks
# 构建合规性约束
rules = []
if "GDPR" in frameworks:
rules.append("- Must obtain explicit consent before processing personal data")
rules.append("- Users have right to data deletion")
if "HIPAA" in frameworks:
rules.append("- Cannot share patient health information without authorization")
rules.append("- Must use secure, encrypted communication")
if industry == "finance":
rules.append("- Cannot provide financial advice without proper disclaimers")
if rules:
compliance_context = f"""Compliance requirements for {jurisdiction}:
{chr(10).join(rules)}"""
# 附加到末尾 - 模型对最后的消息更关注
messages = [
*request.messages,
{"role": "user", "content": compliance_context}
]
request = request.override(messages=messages)
return handler(request)
agent = create_agent(
model="openai:gpt-4o",
tools=[...],
middleware=[inject_compliance_rules],
context_schema=Context
)
复制代码
command语法
它是langgraph的用法
(, 下载次数: 8)
上传
点击文件名下载附件
(, 下载次数: 12)
上传
点击文件名下载附件
from langgraph.graph import StateGraph, END
from langgraph.types import Command
from typing import TypedDict, Literal
# 1. 定义 State
class AgentState(TypedDict):
authenticated: bool
user_role: str
query: str
result: str
# 2. 定义节点函数
def auth_node(state: AgentState) -> Command[Literal["admin_node", "user_node", "END"]]:
"""认证节点:根据角色路由"""
if not state.get("authenticated", False):
return Command(
update={"result": "请先登录"},
goto=END
)
role = state.get("user_role", "user")
if role == "admin":
return Command(goto="admin_node")
else:
return Command(goto="user_node")
def admin_node(state: AgentState) -> Command[Literal["END"]]:
"""管理员节点"""
return Command(
update={"result": f"管理员权限:处理 {state['query']}"},
goto=END
)
def user_node(state: AgentState) -> Command[Literal["END"]]:
"""普通用户节点"""
return Command(
update={"result": f"用户权限:处理 {state['query']}"},
goto=END
)
# 3. 构建图
builder = StateGraph(AgentState)
builder.add_node("auth", auth_node)
builder.add_node("admin", admin_node)
builder.add_node("user", user_node)
builder.set_entry_point("auth")
graph = builder.compile()
# 4. 执行
result = graph.invoke({
"authenticated": True,
"user_role": "admin",
"query": "删除用户数据"
})
print(result["result"]) # 管理员权限:处理删除用户数据
复制代码
(, 下载次数: 11)
上传
点击文件名下载附件
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
agent = create_agent(
model="openai:gpt-4o",
tools=[...],
middleware=[
SummarizationMiddleware(
model="openai:gpt-4o-mini",
max_tokens_before_summary=4000, # 达到 4000 个 token 时触发摘要
messages_to_keep=20, # 摘要后保留最后 20 条消息
),
],
)
复制代码
需要人类审批,配置中断
(, 下载次数: 8)
上传
点击文件名下载附件
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware # [!code highlight]
from langgraph.checkpoint.memory import InMemorySaver # [!code highlight]
agent = create_agent(
model="openai:gpt-4o",
tools=[write_file_tool, execute_sql_tool, read_data_tool],
middleware=[
HumanInTheLoopMiddleware( # [!code highlight]
interrupt_on={
"write_file": True, # 允许所有决策(批准、编辑、拒绝)
"execute_sql": {"allowed_decisions": ["approve", "reject"]}, # 不允许编辑
# 安全操作,无需批准
"read_data": False,
},
# 中断消息的前缀 - 与工具名称和参数结合形成完整消息
# 例如, "Tool execution pending approval: execute_sql with query='DELETE FROM...'"
# 单个工具可以通过在其中断配置中指定 "description" 来覆盖此项
description_prefix="Tool execution pending approval",
),
],
# 人在回路需要检查点来处理中断。
# 在生产环境中,请使用持久性检查点,如 AsyncPostgresSaver。
checkpointer=InMemorySaver(), # [!code highlight]
)
复制代码
interrupt_on中的true就是这个工具要人类审核的意思,默认就是每个行为都要人类去审核,可以执行批准,编辑,或者拒绝。
"execute_sql": {"allowed_decisions": ["approve", "reject"]}, 就是不允许编辑,但是执行这个tool前要得到人类的批准或者拒绝才能继续。
false就是不用人类审核。
下面重点说一下edit类型:
(, 下载次数: 15)
上传
点击文件名下载附件
下面是人类的响应的具体消息:
from langgraph.types import Command
# 人在回路利用 LangGraph 的持久化层。
# 您必须提供一个线程 ID (thread ID) 以将执行与对话线程关联起来,
# 从而使对话能够暂停和恢复(这对于人工审查是必需的)。
config = {"configurable": {"thread_id": "some_id"}} # [!code highlight]
# 运行图直到遇到中断。
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "Delete old records from the database",
}
]
},
config=config # [!code highlight]
)
# 中断包含完整的 HITL 请求,带有 action_requests 和 review_configs
print(result['__interrupt__'])
# > [
# > Interrupt(
# > value={
# > 'action_requests': [
# > {
# > 'name': 'execute_sql',
# > 'arguments': {'query': 'DELETE FROM records WHERE created_at < NOW() - INTERVAL \'30 days\';'},
# > 'description': 'Tool execution pending approval\n\nTool: execute_sql\nArgs: {...}'
# > }
# > ],
# > 'review_configs': [
# > {
# > 'action_name': 'execute_sql',
# > 'allowed_decisions': ['approve', 'reject']
# > }
# > ]
# > }
# > )
# > ]
# 以批准决策恢复
agent.invoke(
Command( # [!code highlight]
resume={"decisions": [{"type": "approve"}]} # 或 "edit", "reject" [!code highlight]
), # [!code highlight]
config=config # 相同的线程 ID 以恢复暂停的对话
)
复制代码
假如人类对多个工具生成了多个决策或者审核结果,那就按顺序来。按照调用工具的消息中的顺序。
(, 下载次数: 9)
上传
点击文件名下载附件
多智能体
有两种,一种是把子智能体当成一个tool,一种是交接(这里省略了)。
from langchain.tools import tool
from langchain.agents import create_agent
subagent1 = create_agent(model="...", tools=[...])
@tool(
"subagent1_name",
description="subagent1_description"
)
def call_subagent1(query: str):
result = subagent1.invoke({
"messages": [{"role": "user", "content": query}]
})
return result["messages"][-1].content
agent = create_agent(model="...", tools=[call_subagent1])
复制代码
自定义注入(给子智能体注入主智能体的状态)
example_state_key是自定义的状态的一个扩展的键
from langchain.agents import AgentState
from langchain.tools import tool, ToolRuntime
class CustomState(AgentState):
example_state_key: str
@tool(
"subagent1_name",
description="subagent1_description"
)
def call_subagent1(query: str, runtime: ToolRuntime[None, CustomState]):
# 应用所需的任何逻辑,将消息转换为合适的输入
subagent_input = some_logic(query, runtime.state["messages"])
result = subagent1.invoke({
"messages": subagent_input,
# 您也可以根据需要在此处传递其他状态键。
# 确保在主智能体和子智能体的状态模式中都定义了这些键。
"example_state_key": runtime.state["example_state_key"]
})
return result["messages"][-1].content
复制代码
控制来自子智能体的输出
塑造主智能体从子智能体接收回的内容的两个常见策略:
修改提示
– 优化子智能体的提示,以
准确指定应返回什么
。
当输出不完整、过于冗长或缺少关键细节时很有用。一个常见的失败模式是子智能体执行了工具调用或推理,但
没有将结果包含在最终消息中
。提醒它控制器(和用户)只能看到最终输出,因此所有相关信息都必须包含在其中。
自定义输出格式化
– 在将子智能体的响应返回给主智能体之前,在代码中对其进行调整或丰富。
示例:除了最终文本之外,将特定的状态键传递回主智能体。这需要将结果包装在一个 Command(或等效结构)中,以便您可以将自定义状态与子智能体的响应合并。
from typing import Annotated
from langchain.agents import AgentState
from langchain.tools import InjectedToolCallId
from langgraph.types import Command
@tool(
"subagent1_name",
description="subagent1_description"
)
# 我们需要将 `tool_call_id` 传递给子智能体,以便它可以使用它来响应工具调用结果
def call_subagent1(
query: str,
tool_call_id: Annotated[str, InjectedToolCallId],
# 您需要返回一个 `Command` 对象,才能包含除最终工具调用之外的更多内容
) -> Command:
result = subagent1.invoke({
"messages": [{"role": "user", "content": query}]
})
return Command(update={
# 这是我们传回的示例状态键
"example_state_key": result["example_state_key"],
"messages": [
ToolMessage(
content=result["messages"][-1].content,
# 我们需要包含工具调用 ID,以便它与正确的工具调用匹配
tool_call_id=tool_call_id
)
]
})
复制代码
(, 下载次数: 6)
上传
点击文件名下载附件
(, 下载次数: 11)
上传
点击文件名下载附件
正式进入deepsearch(还没学完)
1、这套项目重点学什么
很多入门项目只做到“模型能回答”。这当然重要,但离一个可交付的智能体应用还差几步。
在「深度研搜」里,重点不是让模型说得更长,而是解决这些更具体的问题:
一个任务什么时候该查网络,什么时候该查数据库,什么时候该查知识库;子智能体的 description、system_prompt、tools 应该怎样分工;上传文件和生成文件怎样按会话隔离,避免不同任务互相覆盖;长任务为什么不能让 HTTP 请求一直等,为什么要用 WebSocket 回传进度;工具在深层调用中怎样拿到当前 thread_id 和 session_dir;最终结果怎样从命令行示例走到前端页面、输出文件和下载入口。
所以,这套项目更像一条工程练习线:从 DeepAgents 的基本能力开始,逐步把子智能体、记忆、中间件、文件工具、后端接口和前端联调串起来。
学完之后,你应该能明白三件事:
DeepAgents 适合处理哪类长任务,和普通工具型 Agent 有什么区别。主智能体、子智能体、工具、上下文、文件系统之间应该怎样分层。一个多智能体能力怎样通过 FastAPI、WebSocket 和前端页面真正交付出去。
2、这个项目最终做成什么样
从用户视角看,它是一个“深度研究助手”。
用户可以提出这样的任务:
结合公开资料、数据库信息和我上传的文档,整理一份电商行业研究报告,并生成 PDF。复制错误已复制系统背后会按需做这些事:判断任务需要哪些信息来源;用 Tavily 查询公开网络资料;用 MySQL 查询结构化数据;用 RAGFlow 查询内部知识库;读取用户本次上传的 PDF、Word、Excel 或 Markdown 文件;汇总资料,判断信息是否足够;生成 Markdown,必要时再转换成 PDF;把执行过程、最终结果和输出文件展示给前端。流式解析:"""
DeepAgents 快速入门:搜索工具 + 流式解析
演示如何使用 stream 逐步读取 DeepAgent 的执行过程
相比 invoke 只拿最终结果,stream 更适合观察 agent 的中间状态
模型决定调用工具、工具返回结果、模型生成最终回答
"""
import os
from typing import Literal
from deepagents import create_deep_agent
from dotenv import find_dotenv, load_dotenv
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from tavily import TavilyClient
# 读取项目根目录中的 .env,示例依赖 LLM_QWEN_MAX 和 TAVILY_API_KEY
load_dotenv(find_dotenv())
llm_name = os.getenv("LLM_QWEN_MAX")
tavily_key = os.getenv("TAVILY_API_KEY")
# Tavily 客户端负责真正的联网搜索,工具函数中会复用这个客户端
tavily_client = TavilyClient(api_key=tavily_key)
@tool
def internet_search(
query: str,
max_results: int = 5,
topic: Literal["news", "finance", "general"] = "general",
include_raw_content: bool = False,
):
"""
互联网搜索工具
DeepAgent 会根据工具描述和参数签名,自动决定是否调用该工具
include_raw_content=False 时返回摘要内容;True 时会尝试返回更完整的网页原文
"""
print(
f"开始调用网络搜索工具,核心参数为:{query},{max_results},{topic},{include_raw_content}"
)
return tavily_client.search(
query=query,
max_results=max_results,
topic=topic,
include_raw_content=include_raw_content,
)
# 使用 OpenAI 兼容接口初始化千问模型
llm = init_chat_model(model=llm_name, model_provider="openai")
# 创建 DeepAgent,模型负责推理和规划,tools 提供可被调用的外部能力
# 当前示例不配置子智能体,重点观察“主智能体 + 搜索工具”的基本流程
deep_agent = create_deep_agent(
model=llm,
tools=[internet_search],
subagents=[],
system_prompt="""
你是一名严谨的研究员,可以使用 internet_search 工具检索网络信息。
请根据检索结果进行归纳、分析和交叉验证,生成一份结构清晰、信息可靠的中文报告。
""",
)
# 流式执行,stream 会在每个图节点完成后产出一个 chunk
# 常见节点包括 model(模型决策或最终回答)和 tools(工具执行结果)
# 流式处理结果
stream = deep_agent.stream(
{
"messages": [
{
"role": "user",
"content": "请查询人工智能和机器人领域的热门新闻信息,并整理为一份简要报告。",
}
]
}
)
# 循环获取块
for chunk in stream:
# chunk 是一个按节点名组织的字典,例如
# {"model": {"messages": [...]}} 或 {"tools": {"messages": [...]}}
for node_name, state in chunk.items():
# DeepAgents 内部中间件也可能产出空状态或非消息状态,这里只解析消息类状态
if not state or "messages" not in state:
continue
messages = state["messages"]
if not messages or not isinstance(messages, list):
continue
# 每个 chunk 的最后一条消息,通常就是这个节点本次产出的核心信息
last_msg = messages[-1]
if node_name == "model":
# 情况一:模型决定调用工具或子智能体
# model 节点有两类重点事件
# 1. tool_calls 非空,模型决定下一步调用工具或子智能体
# 2. content 非空,模型已经生成最终回答
if last_msg.tool_calls:
for tool_call in last_msg.tool_calls:
if tool_call["name"] == "task":
print(
f"【大模型】决定调用子智能体:{tool_call['args']['subagent_type']}"
)
else:
print(
f"【大模型】决定调用工具:{tool_call['name']} 传入的参数:{tool_call['args']}"
)
# 情况二:模型生成最终结果
elif last_msg.content:
print(f"【大模型】最终执行的结果:{last_msg.content}")
# 情况三:工具执行完成,返回结果
elif node_name == "tools":
# tools 节点返回的是具体工具的执行结果,通常可以推送给前端展示执行进度
tool_return_result = last_msg.content[:100] + "..."
tool_name = last_msg.name
print(f"【agent】调用了{tool_name}工具,返回的结果为:{tool_return_result}")
复制代码
子agent:
https://didilili.github.io/ai-agents-from-zero/#/%E5%AE%9E%E6%88%98%E9%A1%B9%E7%9B%AE-%E6%B7%B1%E5%BA%A6%E7%A0%94%E6%90%9C/3-%E5%AD%90%E6%99%BA%E8%83%BD%E4%BD%93%E8%BF%9B%E9%98%B6%E4%B8%8E%E5%BC%82%E6%AD%A5%E6%89%A7%E8%A1%8C
原文地址:https://blog.csdn.net/weixin_51981189/article/details/161226606
欢迎光临 AI创想 (https://llms-ai.com/)
Powered by Discuz! X3.4