AI创想

标题: deepsearch笔记 [打印本页]

作者: AI小编    时间: 2026-7-7 07:20
标题: deepsearch笔记
作者:后季暖
前面是一些py基础的笔记。

1.pydantic基础语法

pydantic的=,和python的=不一样,这里的=都是默认值是什么的意思!
pydantic的...是必填的意思。这里...表示此字段必填。
  1. class Location(BaseModel):
  2.     longitude: float = Field(...,description="经度")
  3.     latitude: float = Field(...,description="纬度")
复制代码
2.pydantic的验证功能
  1. from pydantic import field_validator
  2. class WeatherInfo(BaseModel):
  3.     temperature: int
  4.    
  5.     @field_validator('temperature',mode='before')
  6.     def parse_temperature(cls,v):
  7.         """解析温度字符串:"16°C" -> 16"""
  8.         if isinstance(v,str):
  9.             v = v.replace('°C','').replace('℃','').strip()
  10.             return int(v)
  11.         return v
复制代码
注意,验证器只在创建对象时执行,创建后修改属性不会触发验证:
  1. # ✅ 创建时经过验证
  2. product = Product(discount="80%")  # discount = 0.8 (float)
  3. # ❌ 直接赋值绕过验证
  4. product.discount = "白嫖"  # 不会报错!不会触发 parse_discount
  5. print(product.discount)    # "白嫖" (字符串!)
复制代码
这就是为什么 Pydantic 官方建议把模型对象当作不可变对象来用,或者开启验证:
  1. # 如果需要修改后也验证,用 model_config
  2. class Product(BaseModel):
  3.     model_config = {'validate_assignment': True}  # 属性赋值时也验证
  4.    
  5.     discount: float
  6.    
  7.     @field_validator('discount', mode='before')
  8.     def parse_discount(cls, v):
  9.         if isinstance(v, str) and '%' in v:
  10.             return float(v.replace('%', '')) / 100
  11.         return v
  12. # 现在直接赋值也会触发验证
  13. product = Product(discount=0.9)
  14. product.discount = "80%"  # ✅ 自动转换,discount = 0.8
  15. product.discount = "白嫖"  # ❌ ValidationError
复制代码
3.items() 就是 Python 字典(dict)里的一个功能

最简单的例子:
  1. # 一个字典
  2. person = {
  3.     "name": "小明",
  4.     "age": 20,
  5.     "city": "北京"
  6. }
  7. # 用 items() 遍历
  8. for key, value in person.items():
  9.     print("键:", key, " 值:", value)
复制代码
4.缓存命中

提示缓存

许多提供商提供提示缓存功能,以减少对相同令牌重复处理的延迟和成本。这些功能可以是隐式显式
警告
提示缓存通常仅在超过最小输入令牌阈值时才会启用。请参阅提供商页面以了解详细信息。
缓存使用情况将反映在模型响应的使用元数据中。
(, 下载次数: 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解释,下面是示例:
  1. from langchain.messages import ToolMessage
  2. # 发送给模型
  3. message_content = "It was the best of times, it was the worst of times."
  4. # 下游可用的 artifact
  5. artifact = {"document_id": "doc_123", "page": 0}
  6. tool_message = ToolMessage(
  7.     content=message_content,
  8.     tool_call_id="call_123",
  9.     name="search_books",
  10.     artifact=artifact,
  11. )
复制代码
其实这段代码的真实执行顺序:
(, 下载次数: 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 限制。这个中间件的做法是:当消息太长时,用一个小模型生成摘要,用摘要替代旧消息
  1. from langchain.agents import create_agent
  2. from langchain.agents.middleware import SummarizationMiddleware
  3. from langgraph.checkpoint.memory import InMemorySaver
  4. from langchain_core.runnables import RunnableConfig
  5. checkpointer = InMemorySaver()
  6. agent = create_agent(
  7.     model="openai:gpt-4o",
  8.     tools=[],
  9.     middleware=[
  10.         SummarizationMiddleware(
  11.             model="openai:gpt-4o-mini",
  12.             max_tokens_before_summary=4000,  # Trigger summarization at 4000 tokens
  13.             messages_to_keep=20,  # Keep last 20 messages after summary
  14.         )
  15.     ],
  16.     checkpointer=checkpointer,
  17. )
  18. config: RunnableConfig = {"configurable": {"thread_id": "1"}}
  19. agent.invoke({"messages": "hi, my name is bob"}, config)
  20. agent.invoke({"messages": "write a short poem about cats"}, config)
  21. agent.invoke({"messages": "now do the same but for dogs"}, config)
  22. final_response = agent.invoke({"messages": "what's my name?"}, config)
  23. final_response["messages"][-1].pretty_print()
  24. """
  25. ================================== Ai Message ==================================
  26. Your name is Bob!
  27. """
复制代码
在工具中读取短期记忆 (Read short-term memory in a tool)

使用 ToolRuntime 参数在工具中访问短期记忆(状态)。
tool_runtime 参数对工具签名是隐藏的(因此模型看不到它),但工具可以通过它访问状态。
这里的短期记忆其实就是状态,下面代码是我自定义了一个状态CustomState。
然后runtime.state就能获取到这个状态。如果你不定义自定义的状态,默认也有一个的。
  1. from langchain.agents import create_agent, AgentState
  2. from langchain.tools import tool, ToolRuntime
  3. class CustomState(AgentState):
  4.     user_id: str
  5. @tool
  6. def get_user_info(
  7.     runtime: ToolRuntime
  8. ) -> str:
  9.     """Look up user info."""
  10.     user_id = runtime.state["user_id"]
  11.     return "User is John Smith" if user_id == "user_123" else "Unknown user"
  12. agent = create_agent(
  13.     model="openai:gpt-5-nano",
  14.     tools=[get_user_info],
  15.     state_schema=CustomState,
  16. )
  17. result = agent.invoke({
  18.     "messages": "look up user information",
  19.     "user_id": "user_123"
  20. })
  21. print(result["messages"][-1].content)
  22. # > User is John Smith.
复制代码
下面两个是完全不一样的。有的时候你自定义state,有时候context,有时候一起都要自定义。
(, 下载次数: 14)


(, 下载次数: 11)


(, 下载次数: 17)


流式传输
  1. from langchain.agents import create_agent
  2. def get_weather(city: str) -> str:
  3.     """获取给定城市的天气。"""
  4.     return f"It's always sunny in {city}!"
  5. agent = create_agent(
  6.     model="openai:gpt-5-nano",
  7.     tools=[get_weather],
  8. )
  9. for chunk in agent.stream(  # [!code highlight]
  10.     {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
  11.     stream_mode="updates",
  12. ):
  13.     for step, data in chunk.items():
  14.         print(f"step: {step}")
  15.         print(f"content: {data['messages'][-1].content_blocks}")
复制代码
(, 下载次数: 8)


chunk的结构:
  1. # chunk 的内容示例:
  2. {
  3.     "model": {                          # 步骤名(字符串键)
  4.         "messages": [                   # 该步骤的更新数据(字典值)
  5.             AIMessage(
  6.                 content_blocks=[...]    # 模型生成的工具调用或文本
  7.             )
  8.         ]
  9.     }
  10. }
  11. # 或者
  12. {
  13.     "tools": {                          # 步骤名
  14.         "messages": [                   # 更新数据
  15.             ToolMessage(
  16.                 content="It's always sunny in SF!",
  17.                 tool_call_id="call_123"
  18.             )
  19.         ]
  20.     }
  21. }
复制代码
这段代码展示了如何在工具执行过程中自定义流式输出,让你能实时推送工具内部的进度信息。
核心机制:get_stream_writer()
  1. from langchain.agents import create_agent
  2. from langgraph.config import get_stream_writer  # [!code highlight]
  3. def get_weather(city: str) -> str:
  4.     """获取给定城市的天气。"""
  5.     writer = get_stream_writer()  # [!code highlight]
  6.     # 流式传输任何任意数据
  7.     writer(f"Looking up data for city: {city}")
  8.     writer(f"Acquired data for city: {city}")
  9.     return f"It's always sunny in {city}!"
  10. agent = create_agent(
  11.     model="anthropic:claude-sonnet-4-5",
  12.     tools=[get_weather],
  13. )
  14. for chunk in agent.stream(
  15.     {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
  16.     stream_mode="custom"  # [!code highlight]
  17. ):
  18.     print(chunk)
复制代码
(, 下载次数: 14)


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

(, 下载次数: 9)


(, 下载次数: 12)


(, 下载次数: 16)


(, 下载次数: 14)


(, 下载次数: 6)


(, 下载次数: 10)


(, 下载次数: 8)


Tool Strategy异常处理

(, 下载次数: 11)


守卫

人工审核,执行工具前会先问用户,得到批准才执行:
  1. from langchain.agents import create_agent
  2. from langchain.agents.middleware import HumanInTheLoopMiddleware
  3. from langgraph.checkpoint.memory import InMemorySaver
  4. from langgraph.types import Command
  5. agent = create_agent(
  6.     model="openai:gpt-4o",
  7.     tools=[search_tool, send_email_tool, delete_database_tool],
  8.     middleware=[
  9.         HumanInTheLoopMiddleware(
  10.             interrupt_on={
  11.                 # 要求批准敏感操作
  12.                 "send_email": True,
  13.                 "delete_database": True,
  14.                 # 自动批准安全操作
  15.                 "search": False,
  16.             }
  17.         ),
  18.     ],
  19.     # 在中断期间持久化状态
  20.     checkpointer=InMemorySaver(),
  21. )
  22. # 人工审核需要一个线程 ID 来进行持久化
  23. config = {"configurable": {"thread_id": "some_id"}}
  24. # 在执行敏感工具之前,智能体将暂停并等待批准
  25. result = agent.invoke(
  26.     {"messages": [{"role": "user", "content": "Send an email to the team"}]},
  27.     config=config
  28. )
  29. result = agent.invoke(
  30.     Command(resume={"decisions": [{"type": "approve"}]}),
  31.     config=config # 相同的线程 ID 以恢复暂停的对话
  32. )
复制代码
(, 下载次数: 9)


request、runtime、context等的关系:

(, 下载次数: 9)


state只限于当前会话,但是store它可以跨会话的。所以store用来当长期记忆。
(, 下载次数: 14)


context常见用法的示例:根据用户的管辖范围,从运行时上下文中注入合规规则:
  1. from dataclasses import dataclass
  2. from langchain.agents import create_agent
  3. from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
  4. from typing import Callable
  5. @dataclass
  6. class Context:
  7.     user_jurisdiction: str
  8.     industry: str
  9.     compliance_frameworks: list[str]
  10. @wrap_model_call
  11. def inject_compliance_rules(
  12.     request: ModelRequest,
  13.     handler: Callable[[ModelRequest], ModelResponse]
  14. ) -> ModelResponse:
  15.     """Inject compliance constraints from Runtime Context."""
  16.     # 从 Runtime Context 读取:获取合规性要求
  17.     jurisdiction = request.runtime.context.user_jurisdiction
  18.     industry = request.runtime.context.industry
  19.     frameworks = request.runtime.context.compliance_frameworks
  20.     # 构建合规性约束
  21.     rules = []
  22.     if "GDPR" in frameworks:
  23.         rules.append("- Must obtain explicit consent before processing personal data")
  24.         rules.append("- Users have right to data deletion")
  25.     if "HIPAA" in frameworks:
  26.         rules.append("- Cannot share patient health information without authorization")
  27.         rules.append("- Must use secure, encrypted communication")
  28.     if industry == "finance":
  29.         rules.append("- Cannot provide financial advice without proper disclaimers")
  30.     if rules:
  31.         compliance_context = f"""Compliance requirements for {jurisdiction}:
  32.     {chr(10).join(rules)}"""
  33.         # 附加到末尾 - 模型对最后的消息更关注
  34.         messages = [
  35.             *request.messages,
  36.             {"role": "user", "content": compliance_context}
  37.         ]
  38.         request = request.override(messages=messages)
  39.     return handler(request)
  40. agent = create_agent(
  41.     model="openai:gpt-4o",
  42.     tools=[...],
  43.     middleware=[inject_compliance_rules],
  44.     context_schema=Context
  45. )
复制代码
command语法

它是langgraph的用法
(, 下载次数: 8)


(, 下载次数: 12)

  1. from langgraph.graph import StateGraph, END
  2. from langgraph.types import Command
  3. from typing import TypedDict, Literal
  4. # 1. 定义 State
  5. class AgentState(TypedDict):
  6.     authenticated: bool
  7.     user_role: str
  8.     query: str
  9.     result: str
  10. # 2. 定义节点函数
  11. def auth_node(state: AgentState) -> Command[Literal["admin_node", "user_node", "END"]]:
  12.     """认证节点:根据角色路由"""
  13.     if not state.get("authenticated", False):
  14.         return Command(
  15.             update={"result": "请先登录"},
  16.             goto=END
  17.         )
  18.    
  19.     role = state.get("user_role", "user")
  20.     if role == "admin":
  21.         return Command(goto="admin_node")
  22.     else:
  23.         return Command(goto="user_node")
  24. def admin_node(state: AgentState) -> Command[Literal["END"]]:
  25.     """管理员节点"""
  26.     return Command(
  27.         update={"result": f"管理员权限:处理 {state['query']}"},
  28.         goto=END
  29.     )
  30. def user_node(state: AgentState) -> Command[Literal["END"]]:
  31.     """普通用户节点"""
  32.     return Command(
  33.         update={"result": f"用户权限:处理 {state['query']}"},
  34.         goto=END
  35.     )
  36. # 3. 构建图
  37. builder = StateGraph(AgentState)
  38. builder.add_node("auth", auth_node)
  39. builder.add_node("admin", admin_node)
  40. builder.add_node("user", user_node)
  41. builder.set_entry_point("auth")
  42. graph = builder.compile()
  43. # 4. 执行
  44. result = graph.invoke({
  45.     "authenticated": True,
  46.     "user_role": "admin",
  47.     "query": "删除用户数据"
  48. })
  49. print(result["result"])  # 管理员权限:处理删除用户数据
复制代码
(, 下载次数: 11)

  1. from langchain.agents import create_agent
  2. from langchain.agents.middleware import SummarizationMiddleware
  3. agent = create_agent(
  4.     model="openai:gpt-4o",
  5.     tools=[...],
  6.     middleware=[
  7.         SummarizationMiddleware(
  8.             model="openai:gpt-4o-mini",
  9.             max_tokens_before_summary=4000,  # 达到 4000 个 token 时触发摘要
  10.             messages_to_keep=20,  # 摘要后保留最后 20 条消息
  11.         ),
  12.     ],
  13. )
复制代码
需要人类审批,配置中断

(, 下载次数: 8)

  1. from langchain.agents import create_agent
  2. from langchain.agents.middleware import HumanInTheLoopMiddleware # [!code highlight]
  3. from langgraph.checkpoint.memory import InMemorySaver # [!code highlight]
  4. agent = create_agent(
  5.     model="openai:gpt-4o",
  6.     tools=[write_file_tool, execute_sql_tool, read_data_tool],
  7.     middleware=[
  8.         HumanInTheLoopMiddleware( # [!code highlight]
  9.             interrupt_on={
  10.                 "write_file": True,  # 允许所有决策(批准、编辑、拒绝)
  11.                 "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
  12.                 # 安全操作,无需批准
  13.                 "read_data": False,
  14.             },
  15.             # 中断消息的前缀 - 与工具名称和参数结合形成完整消息
  16.             # 例如, "Tool execution pending approval: execute_sql with query='DELETE FROM...'"
  17.             # 单个工具可以通过在其中断配置中指定 "description" 来覆盖此项
  18.             description_prefix="Tool execution pending approval",
  19.         ),
  20.     ],
  21.     # 人在回路需要检查点来处理中断。
  22.     # 在生产环境中,请使用持久性检查点,如 AsyncPostgresSaver。
  23.     checkpointer=InMemorySaver(),  # [!code highlight]
  24. )
复制代码
interrupt_on中的true就是这个工具要人类审核的意思,默认就是每个行为都要人类去审核,可以执行批准,编辑,或者拒绝。
"execute_sql": {"allowed_decisions": ["approve", "reject"]},  就是不允许编辑,但是执行这个tool前要得到人类的批准或者拒绝才能继续。
false就是不用人类审核。
下面重点说一下edit类型:
(, 下载次数: 15)


下面是人类的响应的具体消息:
  1. from langgraph.types import Command
  2. # 人在回路利用 LangGraph 的持久化层。
  3. # 您必须提供一个线程 ID (thread ID) 以将执行与对话线程关联起来,
  4. # 从而使对话能够暂停和恢复(这对于人工审查是必需的)。
  5. config = {"configurable": {"thread_id": "some_id"}} # [!code highlight]
  6. # 运行图直到遇到中断。
  7. result = agent.invoke(
  8.     {
  9.         "messages": [
  10.             {
  11.                 "role": "user",
  12.                 "content": "Delete old records from the database",
  13.             }
  14.         ]
  15.     },
  16.     config=config # [!code highlight]
  17. )
  18. # 中断包含完整的 HITL 请求,带有 action_requests 和 review_configs
  19. print(result['__interrupt__'])
  20. # > [
  21. # >     Interrupt(
  22. # >         value={
  23. # >           'action_requests': [
  24. # >               {
  25. # >                   'name': 'execute_sql',
  26. # >                   'arguments': {'query': 'DELETE FROM records WHERE created_at < NOW() - INTERVAL \'30 days\';'},
  27. # >                   'description': 'Tool execution pending approval\n\nTool: execute_sql\nArgs: {...}'
  28. # >               }
  29. # >            ],
  30. # >            'review_configs': [
  31. # >               {
  32. # >                    'action_name': 'execute_sql',
  33. # >                    'allowed_decisions': ['approve', 'reject']
  34. # >               }
  35. # >            ]
  36. # >         }
  37. # >     )
  38. # > ]
  39. # 以批准决策恢复
  40. agent.invoke(
  41.     Command( # [!code highlight]
  42.         resume={"decisions": [{"type": "approve"}]}  # 或 "edit", "reject" [!code highlight]
  43.     ), # [!code highlight]
  44.     config=config # 相同的线程 ID 以恢复暂停的对话
  45. )
复制代码
假如人类对多个工具生成了多个决策或者审核结果,那就按顺序来。按照调用工具的消息中的顺序。
(, 下载次数: 9)


多智能体

有两种,一种是把子智能体当成一个tool,一种是交接(这里省略了)。
  1. from langchain.tools import tool
  2. from langchain.agents import create_agent
  3. subagent1 = create_agent(model="...", tools=[...])
  4. @tool(
  5.     "subagent1_name",
  6.     description="subagent1_description"
  7. )
  8. def call_subagent1(query: str):
  9.     result = subagent1.invoke({
  10.         "messages": [{"role": "user", "content": query}]
  11.     })
  12.     return result["messages"][-1].content
  13. agent = create_agent(model="...", tools=[call_subagent1])
复制代码
自定义注入(给子智能体注入主智能体的状态)
example_state_key是自定义的状态的一个扩展的键
  1. from langchain.agents import AgentState
  2. from langchain.tools import tool, ToolRuntime
  3. class CustomState(AgentState):
  4.     example_state_key: str
  5. @tool(
  6.     "subagent1_name",
  7.     description="subagent1_description"
  8. )
  9. def call_subagent1(query: str, runtime: ToolRuntime[None, CustomState]):
  10.     # 应用所需的任何逻辑,将消息转换为合适的输入
  11.     subagent_input = some_logic(query, runtime.state["messages"])
  12.     result = subagent1.invoke({
  13.         "messages": subagent_input,
  14.         # 您也可以根据需要在此处传递其他状态键。
  15.         # 确保在主智能体和子智能体的状态模式中都定义了这些键。
  16.         "example_state_key": runtime.state["example_state_key"]
  17.     })
  18.     return result["messages"][-1].content
复制代码
控制来自子智能体的输出

塑造主智能体从子智能体接收回的内容的两个常见策略:
  1. from typing import Annotated
  2. from langchain.agents import AgentState
  3. from langchain.tools import InjectedToolCallId
  4. from langgraph.types import Command
  5. @tool(
  6.     "subagent1_name",
  7.     description="subagent1_description"
  8. )
  9. # 我们需要将 `tool_call_id` 传递给子智能体,以便它可以使用它来响应工具调用结果
  10. def call_subagent1(
  11.     query: str,
  12.     tool_call_id: Annotated[str, InjectedToolCallId],
  13. # 您需要返回一个 `Command` 对象,才能包含除最终工具调用之外的更多内容
  14. ) -> Command:
  15.     result = subagent1.invoke({
  16.         "messages": [{"role": "user", "content": query}]
  17.     })
  18.     return Command(update={
  19.         # 这是我们传回的示例状态键
  20.         "example_state_key": result["example_state_key"],
  21.         "messages": [
  22.             ToolMessage(
  23.                 content=result["messages"][-1].content,
  24.                 # 我们需要包含工具调用 ID,以便它与正确的工具调用匹配
  25.                 tool_call_id=tool_call_id
  26.             )
  27.         ]
  28.     })
复制代码
(, 下载次数: 6)


(, 下载次数: 11)


正式进入deepsearch(还没学完)

1、这套项目重点学什么

很多入门项目只做到“模型能回答”。这当然重要,但离一个可交付的智能体应用还差几步。
在「深度研搜」里,重点不是让模型说得更长,而是解决这些更具体的问题:
所以,这套项目更像一条工程练习线:从 DeepAgents 的基本能力开始,逐步把子智能体、记忆、中间件、文件工具、后端接口和前端联调串起来。
学完之后,你应该能明白三件事:
2、这个项目最终做成什么样

从用户视角看,它是一个“深度研究助手”。
用户可以提出这样的任务:
  1. 结合公开资料、数据库信息和我上传的文档,整理一份电商行业研究报告,并生成 PDF。复制错误已复制系统背后会按需做这些事:判断任务需要哪些信息来源;用 Tavily 查询公开网络资料;用 MySQL 查询结构化数据;用 RAGFlow 查询内部知识库;读取用户本次上传的 PDF、Word、Excel 或 Markdown 文件;汇总资料,判断信息是否足够;生成 Markdown,必要时再转换成 PDF;把执行过程、最终结果和输出文件展示给前端。流式解析:"""
  2. DeepAgents 快速入门:搜索工具 + 流式解析
  3. 演示如何使用 stream 逐步读取 DeepAgent 的执行过程
  4. 相比 invoke 只拿最终结果,stream 更适合观察 agent 的中间状态
  5. 模型决定调用工具、工具返回结果、模型生成最终回答
  6. """
  7. import os
  8. from typing import Literal
  9. from deepagents import create_deep_agent
  10. from dotenv import find_dotenv, load_dotenv
  11. from langchain.chat_models import init_chat_model
  12. from langchain.tools import tool
  13. from tavily import TavilyClient
  14. # 读取项目根目录中的 .env,示例依赖 LLM_QWEN_MAX 和 TAVILY_API_KEY
  15. load_dotenv(find_dotenv())
  16. llm_name = os.getenv("LLM_QWEN_MAX")
  17. tavily_key = os.getenv("TAVILY_API_KEY")
  18. # Tavily 客户端负责真正的联网搜索,工具函数中会复用这个客户端
  19. tavily_client = TavilyClient(api_key=tavily_key)
  20. @tool
  21. def internet_search(
  22.     query: str,
  23.     max_results: int = 5,
  24.     topic: Literal["news", "finance", "general"] = "general",
  25.     include_raw_content: bool = False,
  26. ):
  27.     """
  28.     互联网搜索工具
  29.     DeepAgent 会根据工具描述和参数签名,自动决定是否调用该工具
  30.     include_raw_content=False 时返回摘要内容;True 时会尝试返回更完整的网页原文
  31.     """
  32.     print(
  33.         f"开始调用网络搜索工具,核心参数为:{query},{max_results},{topic},{include_raw_content}"
  34.     )
  35.     return tavily_client.search(
  36.         query=query,
  37.         max_results=max_results,
  38.         topic=topic,
  39.         include_raw_content=include_raw_content,
  40.     )
  41. # 使用 OpenAI 兼容接口初始化千问模型
  42. llm = init_chat_model(model=llm_name, model_provider="openai")
  43. # 创建 DeepAgent,模型负责推理和规划,tools 提供可被调用的外部能力
  44. # 当前示例不配置子智能体,重点观察“主智能体 + 搜索工具”的基本流程
  45. deep_agent = create_deep_agent(
  46.     model=llm,
  47.     tools=[internet_search],
  48.     subagents=[],
  49.     system_prompt="""
  50.     你是一名严谨的研究员,可以使用 internet_search 工具检索网络信息。
  51.     请根据检索结果进行归纳、分析和交叉验证,生成一份结构清晰、信息可靠的中文报告。
  52.     """,
  53. )
  54. # 流式执行,stream 会在每个图节点完成后产出一个 chunk
  55. # 常见节点包括 model(模型决策或最终回答)和 tools(工具执行结果)
  56. # 流式处理结果
  57. stream = deep_agent.stream(
  58.     {
  59.         "messages": [
  60.             {
  61.                 "role": "user",
  62.                 "content": "请查询人工智能和机器人领域的热门新闻信息,并整理为一份简要报告。",
  63.             }
  64.         ]
  65.     }
  66. )
  67. # 循环获取块
  68. for chunk in stream:
  69.     # chunk 是一个按节点名组织的字典,例如
  70.     # {"model": {"messages": [...]}} 或 {"tools": {"messages": [...]}}
  71.     for node_name, state in chunk.items():
  72.         # DeepAgents 内部中间件也可能产出空状态或非消息状态,这里只解析消息类状态
  73.         if not state or "messages" not in state:
  74.             continue
  75.         messages = state["messages"]
  76.         if not messages or not isinstance(messages, list):
  77.             continue
  78.         # 每个 chunk 的最后一条消息,通常就是这个节点本次产出的核心信息
  79.         last_msg = messages[-1]
  80.         if node_name == "model":
  81.             # 情况一:模型决定调用工具或子智能体
  82.             # model 节点有两类重点事件
  83.             # 1. tool_calls 非空,模型决定下一步调用工具或子智能体
  84.             # 2. content 非空,模型已经生成最终回答
  85.             if last_msg.tool_calls:
  86.                 for tool_call in last_msg.tool_calls:
  87.                     if tool_call["name"] == "task":
  88.                         print(
  89.                             f"【大模型】决定调用子智能体:{tool_call['args']['subagent_type']}"
  90.                         )
  91.                     else:
  92.                         print(
  93.                             f"【大模型】决定调用工具:{tool_call['name']} 传入的参数:{tool_call['args']}"
  94.                         )
  95.             # 情况二:模型生成最终结果
  96.             elif last_msg.content:
  97.                 print(f"【大模型】最终执行的结果:{last_msg.content}")
  98.         # 情况三:工具执行完成,返回结果
  99.         elif node_name == "tools":
  100.             # tools 节点返回的是具体工具的执行结果,通常可以推送给前端展示执行进度
  101.             tool_return_result = last_msg.content[:100] + "..."
  102.             tool_name = last_msg.name
  103.             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