DeepAgents 上下文管理（全网最细）
·
DeepAgents 上下文管理详解
全景图
智能体的"大脑"由四层上下文组成：
┌─────────────────────────────────────────────────┐
│ 第1层：输入响应层（启动时加载，固定不变） │
│ ┌──────────┬──────────┬──────────┬──────────┐ │
│ │ system_ │ memory │ skills │ @tool │ │
│ │ prompt │ 文件 │ 技能 │ 工具说明 │ │
│ └──────────┴──────────┴──────────┴──────────┘ │
├─────────────────────────────────────────────────┤
│ 第2层：运行时上下文（每次调用动态传入） │
│ ┌────────────────────────────────────────────┐ │
│ │ context_schema 定义 → invoke 时传入 context │ │
│ └────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────┤
│ 第3层：子代理隔离（自动隔离，不可控） │
│ ┌──────────────┬──────────────┐ │
│ │ 子智能体 A │ 子智能体 B │ 互不可见 │
│ └──────────────┴──────────────┘ │
├─────────────────────────────────────────────────┤
│ 第4层：长短期记忆 │
│ ┌──────────────────┬───────────────────────┐ │
│ │ 短期：checkpointer│ 长期：store + backend │ │
│ │ 同一会话内 │ 跨会话 │ │
│ └──────────────────┴───────────────────────┘ │
└─────────────────────────────────────────────────┘
第1层：输入响应层（启动时加载，固定不变）
这一层的内容在 create_deep_agent() 时就确定了，之后不会变。
1.1 system_prompt —— 身份定义
agent = create_deep_agent(
model=llm,
system_prompt="""
你是一个旅行规划师。
## 角色
- 你是专业的旅行顾问
- 你熟悉全球各大旅游城市
## 行为准则
- 回答必须使用中文
- 涉及金额保留两位小数
- 不要编造数据，不确定就说"不确定"
## 工作流程
1. 先了解用户需求
2. 查询天气
3. 推荐景点
4. 计算预算
5. 生成旅行计划
## 边界
- 不要帮用户订机票
- 不要处理签证问题
"""
)
system_prompt 的三个作用：
| 作用 | 说明 |
|---|---|
| 身份定义 | 告诉 AI “你是谁” |
| 行为准则 | 告诉 AI “怎么做事” |
| 边界限制 | 告诉 AI “什么不能做” |
1.2 memory —— 总是被读取的记忆文件
memory 是独立的 markdown 文件，每次执行都会被加载到上下文中。
agent = create_deep_agent(
model=llm,
memory=[
"/project/AGENTS.md", # 项目规则
"/project/preferences.md" # 用户偏好
],
backend=file_backend
)
AGENTS.md 示例：
# 项目规则
## 数据规范
- 所有金额使用人民币，保留两位小数
- 日期格式统一使用 YYYY-MM-DD
- 电话号码脱敏显示：138****1234
## 行为规范
- 回答优先使用表格和列表
- 涉及敏感操作必须提示用户确认
- 日志记录所有工具调用
## 禁止事项
- 不要编造数据
- 不要执行 DELETE 操作
- 不要泄露数据库密码
preferences.md 示例：
# 用户偏好
## 语言
- 默认使用中文回答
- 代码注释使用英文
## 输出格式
- 喜欢简洁的回答
- 优先使用表格对比
- 重要信息加粗标注
## 业务偏好
- 优先推荐性价比高的方案
- 预算超支时自动推荐替代方案
memory 和 system_prompt 的区别：
| 特性 | system_prompt | memory |
|---|---|---|
| 定义位置 | 代码里 | 外部文件 |
| 内容量 | 通常较短 | 可以很长 |
| 修改方式 | 改代码 | 改文件，不改代码 |
| 加载时机 | 启动时 | 启动时 |
| 适用场景 | 核心身份和规则 | 详细的规范和偏好 |
1.3 skills —— 渐进式加载
skills 不会全部加载，只在任务相关时才读取对应的 SKILL.md。
agent = create_deep_agent(
model=llm,
skills=["skills"], # 技能目录
backend=file_backend
)
目录结构：
skills/
├── weather_report/
│ └── SKILL.md # 天气报告技能
├── code_review/
│ └── SKILL.md # 代码审查技能
└── data_analysis/
├── SKILL.md # 数据分析技能
└── scripts/
└── chart.py # 附带脚本
1.4 @tool —— 总是被加载的工具说明
@tool
def query_database(sql: str) -> str:
""""执行 SQL 查询语句。""""
pass
这个 docstring 会被自动提取，作为工具说明加载到上下文中。
1.5 四种组件加载方式对比
agent = create_deep_agent(
model=llm,
system_prompt="你是旅行规划师...", # 总是加载
memory=["/project/AGENTS.md"], # 总是加载
skills=["skills"], # 渐进式加载
tools=[query_database, search_web], # 总是加载（docstring）
backend=file_backend
)
| 组件 | 加载时机 | 加载方式 | 内容来源 |
|---|---|---|---|
| system_prompt | 启动时 | 总是加载 | 代码字符串 |
| memory | 启动时 | 总是加载 | 外部 markdown 文件 |
| @tool | 启动时 | 总是加载 | 函数 docstring |
| skills | 启动时 | 渐进式加载 | SKILL.md 文件 |
第2层：运行时上下文（每次调用动态传入）
这是最实用的一层！解决的问题是：工具执行时需要知道"当前用户是谁"。
2.1 典型场景
用户 A（普通用户）调用：查询订单 → 只能查自己的
用户 B（管理员）调用：查询订单 → 可以查所有人的
没有运行时上下文，工具怎么知道"当前是 A 还是 B"？
2.2 四步实现
第一步：定义上下文数据类型
from dataclasses import dataclass
@dataclass
class UserContext:
user_id: str # 用户 ID
username: str # 用户名
role: str # 角色：admin / user
token: str # 登录 token
department: str # 部门
第二步：创建智能体时指定 context_schema
from deepagents import create_deep_agent
agent = create_deep_agent(
model=llm,
tools=[query_orders, delete_record],
system_prompt="你是一个订单管理助手。",
context_schema=UserContext # 告诉智能体：运行时会传入这个类型
)
第三步：执行时传入上下文
# 普通用户调用
result = agent.invoke(
input={"messages": [{"role": "user", "content": "查询我的订单"}]},
config={"configurable": {"thread_id": "thread_1"}},
context=UserContext(
user_id="U001",
username="大风子",
role="user",
token="eyJhbGciOiJI...",
department="技术部"
)
)
# 管理员调用
result = agent.invoke(
input={"messages": [{"role": "user", "content": "查询所有订单"}]},
config={"configurable": {"thread_id": "thread_2"}},
context=UserContext(
user_id="A001",
username="管理员",
role="admin",
token="eyJhbGciOiJI...",
department="管理部"
)
)
第四步：在工具中读取上下文
from deepagents.tools import ToolRuntime
from langchain.tools import tool
@tool
def query_orders(user_filter: str = "") -> str:
""""查询订单信息""""
runtime: ToolRuntime[UserContext] = get_runtime()
ctx = runtime.context
print(f"[工具] 当前用户: {ctx.username} (ID: {ctx.user_id})")
print(f"[工具] 角色: {ctx.role}")
if ctx.role == "admin":
sql = f"SELECT * FROM orders {user_filter}"
else:
sql = f"SELECT * FROM orders WHERE user_id = '{ctx.user_id}'"
return execute_sql(sql)
2.3 完整流程
用户登录 → 拿到 token、user_id、role
|
v
agent.invoke(
input={"messages": [...]},
config={"thread_id": "..."},
context=UserContext(user_id, username, role, token)
)
|
v
智能体收到 messages
|
v
智能体决定调用 query_orders 工具
|
v
query_orders 内部执行：
runtime = get_runtime()
ctx = runtime.context
+-- ctx.role == "admin" -> 查所有订单
+-- ctx.role == "user" -> 只查自己的订单
|
v
返回结果给智能体 -> 整理后返回给用户
2.4 实际应用场景
# 场景1：权限校验
@tool
def delete_record(table: str, record_id: int) -> str:
""""删除记录""""
ctx = get_runtime().context
if ctx.role != "admin":
return "权限不足：只有管理员可以删除记录"
return f"已删除 {table} 表中 ID 为 {record_id} 的记录"
# 场景2：操作审计
@tool
def export_data(query: str) -> str:
""""导出数据""""
ctx = get_runtime().context
log_audit(user=ctx.username, action="export", detail=query)
return execute_export(query)
# 场景3：API 调用鉴权
@tool
def call_external_api(endpoint: str) -> str:
""""调用外部 API""""
ctx = get_runtime().context
headers = {"Authorization": f"Bearer {ctx.token}"}
return requests.get(endpoint, headers=headers).json()
第3层：子代理隔离（自动，不可控）
这一层你不需要手动管理，理解原理就行。
3.1 隔离机制
主智能体
|
+-- 子智能体 A（有自己的上下文）
| - 看不到子智能体 B 的数据
| - 看不到主智能体的运行时上下文
|
+-- 子智能体 B（有自己的上下文）
- 看不到子智能体 A 的数据
- 看不到主智能体的运行时上下文
3.2 为什么隔离？
# 主智能体传任务给子智能体时，只传 messages
# 不传 context、config 等
# 子智能体收到的只有：
{
"messages": [
HumanMessage(content="请帮我查询北京天气")
]
}
# 子智能体拿不到：
# - 主智能体的 context（用户信息）
# - 主智能体的 config（线程信息）
3.3 如果子智能体也需要用户信息怎么办？
方案：把用户信息写进 messages 里传过去。
# 在主智能体的 system_prompt 中引导
system_prompt="""
你是协调员。在委派任务给子智能体时，必须在任务描述中包含：
1. 用户ID
2. 用户角色
3. 具体任务
"""
第4层：长短期记忆
4.1 短期记忆（checkpointer）
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_deep_agent(
model=llm,
tools=[tool1, tool2],
checkpointer=checkpointer,
system_prompt="..."
)
config = {"configurable": {"thread_id": "thread_1"}}
# 第1次调用
agent.invoke({"messages": [{"role": "user", "content": "我叫大风子"}]}, config=config)
# 第2次调用（同一个 thread_id）
result = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config=config)
print(result['messages'][-1].content) # "你叫大风子"
# 不同的 thread_id，互相隔离
config_b = {"configurable": {"thread_id": "thread_2"}}
result = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config=config_b)
# thread_2 不知道 thread_1 的信息
短期记忆的特点：
| 特性 | 说明 |
|---|---|
| 作用域 | 同一个 thread_id 内 |
| 生命周期 | 程序运行期间（内存） |
| 用途 | 多轮对话、HITL 中断恢复 |
| 持久化 | 不持久化，程序关了就没了 |
4.2 长期记忆（store + backend）
from langgraph.store.memory import InMemoryStore
from deepagents.backends import StoreBackend
store = InMemoryStore()
store_backend = StoreBackend(namespace=lambda ctx: ("filesystem",))
agent = create_deep_agent(
model=llm,
store=store,
backend=store_backend,
system_prompt="..."
)
# Thread A 写入记忆
config_a = {"configurable": {"thread_id": "thread_a"}}
agent.invoke({
"messages": [{"role": "user", "content": "我叫大风子，幸运数字是 7，请记住"}]
}, config=config_a)
# Thread B 读取记忆（不同的 thread_id，但共享同一个 store）
config_b = {"configurable": {"thread_id": "thread_b"}}
result = agent.invoke({
"messages": [{"role": "user", "content": "我叫什么？幸运数字是几？"}]
}, config=config_b)
print(result['messages'][-1].content) # "你叫大风子，幸运数字是 7"
长期记忆的特点：
| 特性 | 说明 |
|---|---|
| 作用域 | 跨会话，所有 thread_id 共享 |
| 生命周期 | 持久化（取决于 store 实现） |
| 用途 | 用户偏好、文件存储、知识积累 |
| 持久化 | 可以（Redis、数据库等） |
4.3 长短期记忆对比
| 维度 | 短期记忆 | 长期记忆 |
|---|---|---|
| 实现 | checkpointer | store + backend |
| 作用域 | 同一 thread_id | 跨所有 thread_id |
| 生命周期 | 程序运行期间 | 可持久化 |
| 典型用途 | 多轮对话、HITL | 用户偏好、文件、知识 |
| 数据位置 | 内存 | 内存 / Redis / 数据库 |
四层上下文协作示例
from dataclasses import dataclass
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from pathlib import Path
@dataclass
class UserContext:
user_id: str
username: str
role: str
workspace_dir = Path("./agent_workspace").resolve()
file_backend = FilesystemBackend(root_dir=workspace_dir, virtual_mode=True)
checkpointer = InMemorySaver()
store = InMemoryStore()
agent = create_deep_agent(
model=llm,
tools=[my_tool1, my_tool2], # 第1层：工具说明（总是加载）
system_prompt="你是一个智能助手...", # 第1层：身份定义（总是加载）
memory=["/project/AGENTS.md"], # 第1层：规则文件（总是加载）
skills=["skills"], # 第1层：技能（渐进式加载）
context_schema=UserContext, # 第2层：运行时上下文类型
backend=file_backend, # 第4层：长期记忆后端
checkpointer=checkpointer, # 第4层：短期记忆
store=store # 第4层：长期记忆存储
)
result = agent.invoke(
input={"messages": [{"role": "user", "content": "帮我查询数据"}]},
config={"configurable": {"thread_id": "thread_1"}}, # 第4层：短期记忆
context=UserContext( # 第2层：运行时上下文
user_id="U001",
username="大风子",
role="admin"
)
)
速记口诀
| 层次 | 一句话 | 关键词 |
|---|---|---|
| 第1层：输入响应 | 启动时加载，固定不变 | system_prompt + memory + skills + @tool |
| 第2层：运行时上下文 | 每次调用动态传入 | context_schema + invoke(context=) |
| 第3层：子代理隔离 | 自动隔离，不可控 | 子智能体只收到 messages |
| 第4层：长短期记忆 | 短期同会话，长期跨会话 | checkpointer vs store |
一句话总结：第1层定身份，第2层传身份，第3层隔离身份，第4层记住身份。
更多推荐
22
0- 0
已为社区贡献10条内容
所有评论(0)