deepsearch笔记

AI小编 · 发表于 2026-7-7 07:20:07

作者：后季暖
前面是一些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.缓存命中

提示缓存

许多提供商提供提示缓存功能，以减少对相同令牌重复处理的延迟和成本。这些功能可以是隐式或显式：

隐式提示缓存：

显式缓存：

警告
提示缓存通常仅在超过最小输入令牌阈值时才会启用。请参阅提供商页面以了解详细信息。

缓存使用情况将反映在模型响应的使用元数据中。

显示缓存的优势：

get_usage_metadata_callback可以直接收集你每个模型的token用量。就按照上面的用法就行。不然你只能手动收集：

其实观察上面的callback都是写在一个config里面的，然后config直接在invoke的时候放进去：

文档位置查看有哪些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,
)

复制代码

其实这段代码的真实执行顺序：

也就是说，artifact其实是tool函数内部返回的一个值，捎带出来的。模型不会去看，是给我们用的。
content_blocks

不同模态的content_block不一样，不只是text，还有：

type除了上面的推理，还有下面的各种多模态文件类型：

还有很多就忽略了，甚至还有tool的：

还有比较少见的：

AgentState

所以比如下面这个中间件，用来裁剪消息的，用到的agentstate就是上面的：
文档地址：短期记忆 | LangChain 中文文档

REMOVE_ALL_MESSAGES就是删除所有的消息

@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，有时候一起都要自定义。

流式传输

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}")

复制代码

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)

复制代码

中间件笔记略，用到的时候可以参考：中间件 | LangChain 中文文档
provider strategy和tool strategy区别

Tool Strategy异常处理

守卫

人工审核，执行工具前会先问用户，得到批准才执行：

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 以恢复暂停的对话
)

复制代码

request、runtime、context等的关系：

state只限于当前会话，但是store它可以跨会话的。所以store用来当长期记忆。

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的用法

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"]) # 管理员权限：处理删除用户数据

复制代码

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 条消息
),
],
)

复制代码

需要人类审批，配置中断

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类型：

下面是人类的响应的具体消息：

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 以恢复暂停的对话
)

复制代码

假如人类对多个工具生成了多个决策或者审核结果，那就按顺序来。按照调用工具的消息中的顺序。

多智能体

有两种，一种是把子智能体当成一个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

复制代码

控制来自子智能体的输出

塑造主智能体从子智能体接收回的内容的两个常见策略：

修改提示 – 优化子智能体的提示，以准确指定应返回什么。
自定义输出格式化 – 在将子智能体的响应返回给主智能体之前，在代码中对其进行调整或丰富。

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
)
]
})

复制代码

正式进入deepsearch（还没学完）

1、这套项目重点学什么

很多入门项目只做到“模型能回答”。这当然重要，但离一个可交付的智能体应用还差几步。
在「深度研搜」里，重点不是让模型说得更长，而是解决这些更具体的问题：

所以，这套项目更像一条工程练习线：从 DeepAgents 的基本能力开始，逐步把子智能体、记忆、中间件、文件工具、后端接口和前端联调串起来。
学完之后，你应该能明白三件事：

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

25位大厂高管转战生成式AI创业！吸金猛，不

deepsearch笔记

LangGraph

关于我们

产品与服务

解决方案

产品与服务