deepsearch、langchain笔记

前面是一些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,modebefore) 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(discount80%) # 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, modebefore) def parse_discount(cls, v): if isinstance(v, str) and % in v: return float(v.replace(%, )) / 100 return v # 现在直接赋值也会触发验证 product Product(discount0.9) product.discount 80% # ✅ 自动转换discount 0.8 product.discount 白嫖 # ❌ ValidationError3.items()就是 Python 字典dict里的一个功能最简单的例子# 一个字典 person { name: 小明, age: 20, city: 北京 } # 用 items() 遍历 for key, value in person.items(): print(键:, key, 值:, value)4.缓存命中提示缓存许多提供商提供提示缓存功能以减少对相同令牌重复处理的延迟和成本。这些功能可以是隐式或显式隐式提示缓存如果请求命中缓存提供商将自动传递成本节省。示例OpenAI 和 GeminiGemini 2.5 及以上。显式缓存提供商允许您手动指示缓存点以获得更大控制或保证成本节省。示例ChatOpenAI通过prompt_cache_key、Anthropic 的 AnthropicPromptCachingMiddleware 和 cache_control 选项、AWS Bedrock、Gemini。警告提示缓存通常仅在超过最小输入令牌阈值时才会启用。请参阅提供商页面以了解详细信息。缓存使用情况将反映在模型响应的使用元数据中。显示缓存的优势get_usage_metadata_callback可以直接收集你每个模型的token用量。就按照上面的用法就行。不然你只能手动收集其实观察上面的callback都是写在一个config里面的然后config直接在invoke的时候放进去文档位置查看有哪些configRunnableConfig | langchain_core | LangChain Referencetool 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( contentmessage_content, tool_call_idcall_123, namesearch_books, artifactartifact, )其实这段代码的真实执行顺序也就是说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( modelopenai:gpt-4o, tools[], middleware[ SummarizationMiddleware( modelopenai:gpt-4o-mini, max_tokens_before_summary4000, # Trigger summarization at 4000 tokens messages_to_keep20, # Keep last 20 messages after summary ) ], checkpointercheckpointer, ) 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: whats 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( modelopenai:gpt-5-nano, tools[get_user_info], state_schemaCustomState, ) 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 fIts always sunny in {city}! agent create_agent( modelopenai:gpt-5-nano, tools[get_weather], ) for chunk in agent.stream( # [!code highlight] {messages: [{role: user, content: What is the weather in SF?}]}, stream_modeupdates, ): for step, data in chunk.items(): print(fstep: {step}) print(fcontent: {data[messages][-1].content_blocks})chunk的结构# chunk 的内容示例 { model: { # 步骤名字符串键 messages: [ # 该步骤的更新数据字典值 AIMessage( content_blocks[...] # 模型生成的工具调用或文本 ) ] } } # 或者 { tools: { # 步骤名 messages: [ # 更新数据 ToolMessage( contentIts always sunny in SF!, tool_call_idcall_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(fLooking up data for city: {city}) writer(fAcquired data for city: {city}) return fIts always sunny in {city}! agent create_agent( modelanthropic:claude-sonnet-4-5, tools[get_weather], ) for chunk in agent.stream( {messages: [{role: user, content: What is the weather in SF?}]}, stream_modecustom # [!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( modelopenai:gpt-4o, tools[search_tool, send_email_tool, delete_database_tool], middleware[ HumanInTheLoopMiddleware( interrupt_on{ # 要求批准敏感操作 send_email: True, delete_database: True, # 自动批准安全操作 search: False, } ), ], # 在中断期间持久化状态 checkpointerInMemorySaver(), ) # 人工审核需要一个线程 ID 来进行持久化 config {configurable: {thread_id: some_id}} # 在执行敏感工具之前智能体将暂停并等待批准 result agent.invoke( {messages: [{role: user, content: Send an email to the team}]}, configconfig ) result agent.invoke( Command(resume{decisions: [{type: approve}]}), configconfig # 相同的线程 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 fCompliance requirements for {jurisdiction}: {chr(10).join(rules)} # 附加到末尾 - 模型对最后的消息更关注 messages [ *request.messages, {role: user, content: compliance_context} ] request request.override(messagesmessages) return handler(request) agent create_agent( modelopenai:gpt-4o, tools[...], middleware[inject_compliance_rules], context_schemaContext )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: 请先登录}, gotoEND ) role state.get(user_role, user) if role admin: return Command(gotoadmin_node) else: return Command(gotouser_node) def admin_node(state: AgentState) - Command[Literal[END]]: 管理员节点 return Command( update{result: f管理员权限处理 {state[query]}}, gotoEND ) def user_node(state: AgentState) - Command[Literal[END]]: 普通用户节点 return Command( update{result: f用户权限处理 {state[query]}}, gotoEND ) # 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( modelopenai:gpt-4o, tools[...], middleware[ SummarizationMiddleware( modelopenai:gpt-4o-mini, max_tokens_before_summary4000, # 达到 4000 个 token 时触发摘要 messages_to_keep20, # 摘要后保留最后 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( modelopenai: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 queryDELETE FROM... # 单个工具可以通过在其中断配置中指定 description 来覆盖此项 description_prefixTool execution pending approval, ), ], # 人在回路需要检查点来处理中断。 # 在生产环境中请使用持久性检查点如 AsyncPostgresSaver。 checkpointerInMemorySaver(), # [!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, } ] }, configconfig # [!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] configconfig # 相同的线程 ID 以恢复暂停的对话 )假如人类对多个工具生成了多个决策或者审核结果那就按顺序来。按照调用工具的消息中的顺序。多智能体有两种一种是把子智能体当成一个tool一种是交接这里省略了。from langchain.tools import tool from langchain.agents import create_agent subagent1 create_agent(model..., tools[...]) tool( subagent1_name, descriptionsubagent1_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, descriptionsubagent1_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, descriptionsubagent1_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( contentresult[messages][-1].content, # 我们需要包含工具调用 ID以便它与正确的工具调用匹配 tool_call_idtool_call_id ) ] })正式进入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_keytavily_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_contentFalse 时返回摘要内容True 时会尝试返回更完整的网页原文 print( f开始调用网络搜索工具核心参数为{query},{max_results},{topic},{include_raw_content} ) return tavily_client.search( queryquery, max_resultsmax_results, topictopic, include_raw_contentinclude_raw_content, ) # 使用 OpenAI 兼容接口初始化千问模型 llm init_chat_model(modelllm_name, model_provideropenai) # 创建 DeepAgent模型负责推理和规划tools 提供可被调用的外部能力 # 当前示例不配置子智能体重点观察“主智能体 搜索工具”的基本流程 deep_agent create_deep_agent( modelllm, 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})子agenthttps://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