一、简介

翻了一圈,官网都没有详细介绍create_deep_agent中skils参数具体该怎么用

本文将使用langchain中间件完成skills,一个个人信息查询工具,以及HTML文件助手SKILS

在这里插入图片描述

官方API https://reference.langchain.com/python/deepagents/graph/#deepagents.graph.create_deep_agent

在这里插入图片描述

❤本系列文章,配套项目源码地址❤
https://github.com/wenwenc9/langgraph-tutorial-wenwenc9

在这里插入图片描述

Langgraph系列文章
01|Langgraph | 从入门到实战 | 基础篇
02|Langgraph | 从入门到实战 | workflow与Agent
03|Langgraph | 从入门到实战 | 进阶篇 | 持久化
04|Langgraph | 从入门到实战 | 进阶篇 | 流式传输
05|Langgraph | 从入门到实战 | 进阶篇 | 中断interrupt

langchain的系列文章(相信我把Langchain全部学一遍,你能深入理解AI的开发)
01|LangChain | 从入门到实战-介绍
02|LangChain | 从入门到实战 -六大组件之Models IO
03|LangChain | 从入门到实战 -六大组件之Retrival
04|LangChain | 从入门到实战 -六大组件之Chain
05|LangChain | 从入门到实战 -六大组件之Memory
06|LangChain | 从入门到实战 -六大组件之Agent

二、概要

本文挑选官方文档部分内容讲解,详细内容见官网

https://docs.langchain.com/oss/python/deepagents/overview

在这里插入图片描述

1.1 什么是 Deep Agents?

Deep Agents(深度智能体) 是专门设计用于处理复杂、耗时任务的智能体系统。

复杂任务

Deep Agent

任务规划

工具调用

子代理委托

文件系统管理

任务完成

典型特征:

  • 📊 任务长度约每七个月翻倍
  • 🔧 平均包含 50+ 次不同工具调用
  • ⏱️ 执行时间可能达到数小时甚至数天

典型案例:

  • Claude Code:代码生成和重构
  • Deep Research:深度研究报告
  • Manus:复杂任务自动化

1.2 DeepAgents 核心工具特点

DeepAgents
核心工具

文件系统访问

读取文件

写入文件

编辑文件

文件搜索

Shell工具

执行脚本

命令行操作

规划能力

任务分解

TODO管理

子代理委托

独立上下文

任务隔离

结果反馈

设计理念:

  1. 少量原子工具 + 复杂提示词:不依赖大量不同工具,而是聚焦几个高效的"原子工具"
  2. 充分利用文件系统:分担语言模型上下文窗口压力
  3. 子代理委托:处理 token 密集型任务并隔离上下文

1.3 DeepAgents 与其他框架的关系

DeepAgents框架

预定义工具

倾向性提示词

Skills系统

中间件

LangChain抽象层

Chat Models

Tool Definitions

通用抽象

LangGraph基础设施

Checkpointing

内存管理

人机交互

层级关系:

  • LangGraph:底层运行时环境
  • LangChain:通用抽象层
  • DeepAgents:倾向性智能体框架

1.4 何时使用 Deep Agents?

适用场景:

  • 处理复杂的多步骤任务,需要规划和分解
  • 通过文件系统工具管理大量上下文
  • 委托工作给专门的子代理,用于隔离上下文
  • 需要持久内存,会跨对话和线程

不适用场景:

  • 简单的单次对话
  • 轻量级问答任务
  • 实时响应要求极高的场景

💡 提示:简单的 Agent 建议使用 LangChain Agents

三、环境安装

2.1 安装依赖

from dotenv import load_dotenv, find_dotenv
import os

# 自动向上查找 .env 文件
load_dotenv(find_dotenv())
modelName = os.getenv('OPENAI_MODEL')
TAVILY_API_KEY = os.getenv('TAVILY_API_KEY')
# 安装核心依赖
uv add deepagents tavily-python
uv add deepagents-cli

2.2 注册 API Key

在这里插入图片描述

  • OpenAI API:配置在 .env 文件中

四、快速开始

3.1 创建第一个 Deep Agent

改自官方文档案例:https://docs.langchain.com/oss/python/deepagents/quickstart

这个案例需要tavily 这个网络检索工具

import os
from typing import Literal
from tavily import TavilyClient
from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI

# 初始化 Tavily 客户端
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

# 创建网络检索工具
def internet_search(
    query: str,
    max_results: int = 5,
    topic: Literal["general", "news", "finance"] = "general",
    include_raw_content: bool = False,
):
    """执行网络搜索"""
    return tavily_client.search(
        query,
        max_results=max_results,
        include_raw_content=include_raw_content,
        topic=topic,
    )

# 系统提示词
research_instructions = """你是一位专业的研究者。你的工作是进行全面的研究,然后撰写一份精致的报告。
你可以使用互联网搜索工具作为主要获取信息的手段。

## `internet_search`
使用此工具根据给定查询进行互联网搜索。你可以指定要返回的最大结果数量、主题以及是否包含原始内容。
"""

# 创建 LLM
llm = ChatOpenAI(model=modelName)

# 创建 Deep Agent
agent = create_deep_agent(
    model=llm,
    tools=[internet_search],
    system_prompt=research_instructions
)

3.2 执行任务

result = agent.invoke({
    "messages": [{
        "role": "user", 
        "content": "上海到洛杉矶运输一批服装20gp物流运输解决报告"
    }]
})
for msg in result["messages"]:
    if hasattr(msg, 'pretty_print'):
        msg.pretty_print()

我们查看堆栈内容

  • messages 一个用户问题进来 ,AI自我多轮调用的信息载体
  • files 生成系列文件的指引
  • todos 工作待办内容

由于控制台内容过多,自行到我代码仓库查看,这个任务很好的解决了深度研究的问题,实现的效果内容还挺不错的

这个过程,并且调用了内置的工具,比如write_todoswrite_todosread_filewrite_file

以及tavily 的网络搜索工具 internet_search

在这里插入图片描述

而我们本地目录并没有写入文件,这是怎么因为没有启用backends的参数,后面会提到

https://docs.langchain.com/oss/python/deepagents/backends

在这里插入图片描述

工作流程:

Language Model Internet Search Deep Agent 用户 Language Model Internet Search Deep Agent 用户 loop [多次搜索] 发送任务请求 理解任务 生成执行计划 调用搜索工具 返回搜索结果 分析结果 生成最终报告 返回完整报告

五、skills

关于skills,可以阅读下面资料

想要学习如何开发skills

我当前 deepAgents 版本是 0.3.6 版本

1、创建一个深度代理

使用方法create_deep_agent

https://github.com/langchain-ai/deepagents/blob/deepagents%3D%3D0.3.6/libs/deepagents/deepagents/graph.py#L44

1.1 定义一个查询个人信息工具

import os
import uuid
def query_personal_infos(name: str) -> dict:
    """
    通过姓名查询个人资料工具
    :param name: 姓名
    :return: 个人信息dict
    """

    database = {
        "张伟": {
            "name": "张伟",
            "age": 28,
            "gender": "男",
            "city": "北京",
            "occupation": "软件工程师",
            "company": "字节跳动",
            "education": "清华大学 计算机科学硕士",
            "hobbies": ["编程", "篮球", "阅读"],
            "skills": ["Python", "Java", "React", "Docker"],
            "email": "zhangwei@example.com",
            "phone": "138-0000-1001",
            "bio": "热爱技术,喜欢探索新事物,5年全栈开发经验",
            "avatar": "👨‍💻",
            "salary": "35K",
            "projects": ["电商平台", "数据分析系统", "AI聊天机器人"]
        },
        "李娜": {
            "name": "李娜",
            "age": 26,
            "gender": "女",
            "city": "上海",
            "occupation": "产品经理",
            "company": "阿里巴巴",
            "education": "复旦大学 工商管理本科",
            "hobbies": ["骑行", "摄影", "旅行", "瑜伽"],
            "skills": ["产品设计", "用户研究", "数据分析", "Axure"],
            "email": "lina@example.com",
            "phone": "139-0000-2002",
            "bio": "追求极致用户体验,擅长B端产品设计,3年产品经验",
            "avatar": "👩‍💼",
            "salary": "30K",
            "projects": ["CRM系统", "供应链管理平台", "移动办公App"]
        },
        "王芳": {
            "name": "王芳",
            "age": 32,
            "gender": "女",
            "city": "深圳",
            "occupation": "UI/UX设计师",
            "company": "腾讯",
            "education": "中央美术学院 视觉传达设计硕士",
            "hobbies": ["绘画", "看展", "咖啡", "音乐"],
            "skills": ["Figma", "Sketch", "Photoshop", "Illustrator", "Principle"],
            "email": "wangfang@example.com",
            "phone": "137-0000-3003",
            "bio": "设计驱动创新,关注情感化设计,7年设计经验",
            "avatar": "🎨",
            "salary": "32K",
            "projects": ["微信小程序设计", "企业官网改版", "品牌视觉系统"]
        },
        "刘强": {
            "name": "刘强",
            "age": 35,
            "gender": "男",
            "city": "杭州",
            "occupation": "数据科学家",
            "company": "网易",
            "education": "浙江大学 统计学博士",
            "hobbies": ["跑步", "围棋", "机器学习", "开源贡献"],
            "skills": ["Python", "R", "TensorFlow", "PyTorch", "SQL", "Spark"],
            "email": "liuqiang@example.com",
            "phone": "136-0000-4004",
            "bio": "AI算法专家,专注推荐系统和NLP,10年数据分析经验",
            "avatar": "🔬",
            "salary": "45K",
            "projects": ["推荐引擎优化", "用户画像系统", "智能客服"]
        }
    }

1.2 指派文件系统

# 创建后端(使用当前目录)
from pathlib import Path
current_dir = Path(".").resolve()  # 获取绝对路径
print(f"工作目录: {current_dir}")
from deepagents.backends import FilesystemBackend
backend = FilesystemBackend(root_dir=current_dir, virtual_mode=True)
# backend = FilesystemBackend(root_dir='.', virtual_mode=True)

这里需要提一个问题,使用绝地路径的时候virtual_mode=True

如果你是windows,且出现了错误ValueError: Windows absolute paths are not supported:
在这里插入图片描述
需要修改源码:
.venv/Lib/site-packages/deepagents/backends/filesystem.py

第160行,新增代码

relative_path = relative_path.replace("\\", "/")

要修复 virtual_mode=False,需要:
修改多个文件(skills.py、filesystem.py、可能还有其他)
修改路径验证逻辑
修改路径传递机制
工作量太大,不值得。

我去官方看到看到2025、11、23日,不知道是不是官方的BUG

https://github.com/langchain-ai/deepagents/pull/454

在这里插入图片描述

1.3 使用create_deep_agent

使用短期记忆,还有指定skills参数

from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
from langchain_openai import ChatOpenAI

# 创建 Deep Agent
from deepagents import create_deep_agent
llm = ChatOpenAI(model=modelName)
# 研究指令
research_instructions = """
    你是一个智能助手
"""
agent = create_deep_agent(
    model=llm,
    tools=[query_personal_infos],
    system_prompt=research_instructions,
    # 关键:指定 skills 目录
    skills=[
    # "/.deepagents/skills/",
    "/.deepagents/skills",
    ],
    backend=backend,
    checkpointer=checkpointer,
    # debug=True,  # 开启调试模式查看详细日志
)

同时新建,技能文件夹,我已提交到github了

  • 一个是吹牛技能,用户输入内容,将其内容进行吹牛
  • 另外一个是,查询个人信息的内容,输出保存为html文件
    在这里插入图片描述

1.4 查看是否绑定skills

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
# 执行调用
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "你好,请告诉我你有哪些可用的技能?"
    }]
}, config)

# 检查状态
state = agent.get_state(config)
print(f"状态中的键: {list(state.values.keys())}")

if "skills_metadata" in state.values:
    skills = state.values["skills_metadata"]
    print(f"\n✅ skills_metadata 存在,包含 {len(skills)} 个技能")
    for skill in skills:
        print(f"  - {skill['name']}")
        print(f"    描述: {skill['description']}")
        print(f"    路径: {skill['path']}")
else:
    print("\n❌ skills_metadata 不存在于状态中")
    print("   可能原因:")
    print("   1. before_agent 方法没有被调用")
    print("   2. 技能加载过程中出现异常(被静默忽略)")
    print("   3. 路径配置错误导致找不到技能文件")

# 检查 AI 回复
print("\n" + "=" * 60)
print("步骤4:检查 AI 回复")
print("=" * 60)
print("AI 回复:")
print(result["messages"][-1].content)
print("=" * 60)

如果你出不了,那么考虑考虑 1.2 章节修复一下代码
在这里插入图片描述

2、普通问答

是不会触发skills的

# 测试 read_file 工具是否能读取 SKILL.md
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
test_result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "你好"
    }]
}, config)

# 查看所有消息
for msg in test_result["messages"]:
    if hasattr(msg, 'pretty_print'):
        msg.pretty_print()

在这里插入图片描述

3、使用吹牛技能

在这里插入图片描述

# 测试 read_file 工具是否能读取 SKILL.md
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
test_result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "我可以一口气吃10大碗饭,帮我润色吹牛"
    }]
}, config)

# 查看所有消息
for msg in test_result["messages"]:
    if hasattr(msg, 'pretty_print'):
        msg.pretty_print()

在这里插入图片描述

4、使用html助手

在这里插入图片描述

我们之前创建了一个个人信息查询工具,下面案例执行,会调用多种工具,然后最后调用html助手写入生成美观的html

你可以同时传入多个人名也可以,只不过会慢些

# 测试 read_file 工具是否能读取 SKILL.md
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
test_result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "帮我查询张伟,查询后的信息写入到html文件进行保存在本地"
    }]
}, config)

# 查看所有消息
for msg in test_result["messages"]:
    if hasattr(msg, 'pretty_print'):
        msg.pretty_print()

运行结果,完整过程很多,请看github我提交的jupyter

下面生成了4个todo,且最后一个声明HTML文件保存到本地了
在这里插入图片描述

可以看到写入且帮我保存了文件
在这里插入图片描述

在这里插入图片描述

六、中间件-案例一

此处代码,参考官方文档更改:
https://docs.langchain.com/oss/python/langchain/multi-agent/skills-sql-assistant#view-complete-runnable-script

langchain实现这个技能,本质就是利用中间件的方式,这个案例是将SKILL写成prompt的方式,并未利用内置skills方式

在这里插入图片描述

import os
import uuid
from typing import TypedDict
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.messages import SystemMessage
from langchain.agents.middleware import ModelRequest, ModelResponse, AgentMiddleware
from typing import Callable

# ==================== 个人信息数据库 ====================
def query_personal_infos(name: str) -> dict:
    """
    通过姓名查询个人资料工具
    :param name: 姓名
    :return: 个人信息dict
    """
    database = {
        "张伟": {
            "name": "张伟",
            "age": 28,
            "gender": "男",
            "city": "北京",
            "occupation": "软件工程师",
            "company": "字节跳动",
            "education": "清华大学 计算机科学硕士",
            "hobbies": ["编程", "篮球", "阅读"],
            "skills": ["Python", "Java", "React", "Docker"],
            "email": "zhangwei@example.com",
            "phone": "138-0000-1001",
            "bio": "热爱技术,喜欢探索新事物,5年全栈开发经验",
            "avatar": "👨‍💻",
            "salary": "35K",
            "projects": ["电商平台", "数据分析系统", "AI聊天机器人"]
        },
        "李娜": {
            "name": "李娜",
            "age": 26,
            "gender": "女",
            "city": "上海",
            "occupation": "产品经理",
            "company": "阿里巴巴",
            "education": "复旦大学 工商管理本科",
            "hobbies": ["骑行", "摄影", "旅行", "瑜伽"],
            "skills": ["产品设计", "用户研究", "数据分析", "Axure"],
            "email": "lina@example.com",
            "phone": "139-0000-2002",
            "bio": "追求极致用户体验,擅长B端产品设计,3年产品经验",
            "avatar": "👩‍💼",
            "salary": "30K",
            "projects": ["CRM系统", "供应链管理平台", "移动办公App"]
        },
        "王芳": {
            "name": "王芳",
            "age": 32,
            "gender": "女",
            "city": "深圳",
            "occupation": "UI/UX设计师",
            "company": "腾讯",
            "education": "中央美术学院 视觉传达设计硕士",
            "hobbies": ["绘画", "看展", "咖啡", "音乐"],
            "skills": ["Figma", "Sketch", "Photoshop", "Illustrator", "Principle"],
            "email": "wangfang@example.com",
            "phone": "137-0000-3003",
            "bio": "设计驱动创新,关注情感化设计,7年设计经验",
            "avatar": "🎨",
            "salary": "32K",
            "projects": ["微信小程序设计", "企业官网改版", "品牌视觉系统"]
        },
        "刘强": {
            "name": "刘强",
            "age": 35,
            "gender": "男",
            "city": "杭州",
            "occupation": "数据科学家",
            "company": "网易",
            "education": "浙江大学 统计学博士",
            "hobbies": ["跑步", "围棋", "机器学习", "开源贡献"],
            "skills": ["Python", "R", "TensorFlow", "PyTorch", "SQL", "Spark"],
            "email": "liuqiang@example.com",
            "phone": "136-0000-4004",
            "bio": "AI算法专家,专注推荐系统和NLP,10年数据分析经验",
            "avatar": "🔬",
            "salary": "45K",
            "projects": ["推荐引擎优化", "用户画像系统", "智能客服"]
        }
    }
    return database.get(name, {"error": f"未找到 {name} 的信息"})


# ==================== 技能定义 ====================
class Skill(TypedDict):
    """可以逐步披露给 agent 的技能"""
    name: str
    description: str
    content: str


SKILLS: list[Skill] = [
    {
        "name": "write_interactive_html_report",
        "description": "生成美观的交互式 HTML 个人资料报告",
        "content": """# HTML 报告生成技能

使用 write_file 工具生成包含以下元素的 HTML:
- 渐变背景: linear-gradient(135deg, #667eea 0%, #764ba2 100%)
- 白色卡片容器,圆角 20px
- 头部显示 avatar emoji 和姓名
- 网格布局展示基本信息
- 技能、爱好、项目使用标签样式
- 响应式设计

文件名格式: {姓名}_profile.html
""",
    },
]


# ==================== 技能加载工具 ====================
@tool
def load_skill(skill_name: str) -> str:
    """
    加载技能的完整内容

    Args:
        skill_name: 技能名称
    """
    for skill in SKILLS:
        if skill["name"] == skill_name:
            return f"已加载技能: {skill_name}\n\n{skill['content']}"

    available = ", ".join(s["name"] for s in SKILLS)
    return f"未找到技能 '{skill_name}'。可用技能: {available}"


# ==================== 技能中间件 ====================
class SkillMiddleware(AgentMiddleware):
    """将技能描述注入系统提示词的中间件"""

    tools = [load_skill]

    def __init__(self):
        skills_list = []
        for skill in SKILLS:
            skills_list.append(f"- **{skill['name']}**: {skill['description']}")
        self.skills_prompt = "\n".join(skills_list)

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        skills_addendum = (
            f"\n\n## 可用技能\n\n{self.skills_prompt}\n\n"
            "使用 load_skill 工具获取详细信息。"
        )

        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": skills_addendum}
        ]
        new_system_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        return handler(modified_request)


# ==================== 创建 Agent ====================
research_instructions = """
你是个人资料处理助手。

使用 query_personal_infos 查询信息(一次一个姓名)。
使用 load_skill 加载技能后,用 write_file 生成 HTML 文件。
"""

llm = ChatOpenAI(model=modelName)
backend = FilesystemBackend(root_dir=".", virtual_mode=True)

agent = create_deep_agent(
    model=llm,
    tools=[query_personal_infos],
    middleware=[SkillMiddleware()],
    system_prompt=research_instructions,
    backend=backend
)

调用代码,这次我们换王芳

thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "帮我查询王芳的信息并生成 HTML 文件"
            }
        ]
    },
    config
)

for message in result["messages"]:
    if hasattr(message, 'pretty_print'):
        message.pretty_print()
    else:
        print(f"{message.type}: {message.content}")

在这里插入图片描述
输出效果
在这里插入图片描述

七、中间件-案例二

这里我们自己实现SkillLoader 完全从技能文件目录加载文件

同时可以利用我下面的代码进行debug学习代码堆栈如何运行

import os
import uuid
from typing import TypedDict, Optional
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.messages import SystemMessage
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from typing import Callable
import yaml
import re

# ==================== 个人信息数据库 ====================
def query_personal_infos(name: str) -> dict:
    """
    通过姓名查询个人资料工具
    :param name: 姓名
    :return: 个人信息dict
    """
    database = {
        "张伟": {
            "name": "张伟",
            "age": 28,
            "gender": "男",
            "city": "北京",
            "occupation": "软件工程师",
            "company": "字节跳动",
            "education": "清华大学 计算机科学硕士",
            "hobbies": ["编程", "篮球", "阅读"],
            "skills": ["Python", "Java", "React", "Docker"],
            "email": "zhangwei@example.com",
            "phone": "138-0000-1001",
            "bio": "热爱技术,喜欢探索新事物,5年全栈开发经验",
            "avatar": "👨‍💻",
            "salary": "35K",
            "projects": ["电商平台", "数据分析系统", "AI聊天机器人"]
        },
        "李娜": {
            "name": "李娜",
            "age": 26,
            "gender": "女",
            "city": "上海",
            "occupation": "产品经理",
            "company": "阿里巴巴",
            "education": "复旦大学 工商管理本科",
            "hobbies": ["骑行", "摄影", "旅行", "瑜伽"],
            "skills": ["产品设计", "用户研究", "数据分析", "Axure"],
            "email": "lina@example.com",
            "phone": "139-0000-2002",
            "bio": "追求极致用户体验,擅长B端产品设计,3年产品经验",
            "avatar": "👩‍💼",
            "salary": "30K",
            "projects": ["CRM系统", "供应链管理平台", "移动办公App"]
        },
        "王芳": {
            "name": "王芳",
            "age": 32,
            "gender": "女",
            "city": "深圳",
            "occupation": "UI/UX设计师",
            "company": "腾讯",
            "education": "中央美术学院 视觉传达设计硕士",
            "hobbies": ["绘画", "看展", "咖啡", "音乐"],
            "skills": ["Figma", "Sketch", "Photoshop", "Illustrator", "Principle"],
            "email": "wangfang@example.com",
            "phone": "137-0000-3003",
            "bio": "设计驱动创新,关注情感化设计,7年设计经验",
            "avatar": "🎨",
            "salary": "32K",
            "projects": ["微信小程序设计", "企业官网改版", "品牌视觉系统"]
        },
        "刘强": {
            "name": "刘强",
            "age": 35,
            "gender": "男",
            "city": "杭州",
            "occupation": "数据科学家",
            "company": "网易",
            "education": "浙江大学 统计学博士",
            "hobbies": ["跑步", "围棋", "机器学习", "开源贡献"],
            "skills": ["Python", "R", "TensorFlow", "PyTorch", "SQL", "Spark"],
            "email": "liuqiang@example.com",
            "phone": "136-0000-4004",
            "bio": "AI算法专家,专注推荐系统和NLP,10年数据分析经验",
            "avatar": "🔬",
            "salary": "45K",
            "projects": ["推荐引擎优化", "用户画像系统", "智能客服"]
        }
    }
    return database.get(name, {"error": f"未找到 {name} 的信息"})


# ==================== 工具函数 ====================
def parse_yaml_frontmatter(content: str) -> dict:
    """
    解析 Markdown 文件的 YAML 前置元数据

    格式:
    ---
    name: skill_name
    description: skill description
    ---

    # 正文内容
    """
    pattern = r'^---\s*\n(.*?)\n---\s*\n(.*)$'
    match = re.match(pattern, content, re.DOTALL)

    if match:
        yaml_content = match.group(1)
        markdown_content = match.group(2)
        metadata = yaml.safe_load(yaml_content)
        metadata['content'] = markdown_content.strip()
        return metadata
    else:
        # 如果没有 YAML 前置元数据,返回空元数据
        return {
            'name': 'unknown',
            'description': 'No description',
            'content': content
        }


# ==================== 技能加载工具 ====================
class SkillLoader:
    """技能加载器,负责从目录加载技能"""

    def __init__(self, skills_dirs: list[str]):
        """
        初始化技能加载器

        Args:
            skills_dirs: 技能目录列表
        """
        self.skills_dirs = skills_dirs
        self.skills_cache = {}  # 缓存已加载的技能
        self._load_all_skills()

    def _load_all_skills(self):
        """扫描所有技能目录,加载技能元数据"""
        for skills_dir in self.skills_dirs:
            if not os.path.exists(skills_dir):
                print(f"⚠️  技能目录不存在: {skills_dir}")
                continue

            # 遍历目录中的所有子文件夹
            for item in os.listdir(skills_dir):
                skill_path = os.path.join(skills_dir, item)

                # 检查是否为目录
                if os.path.isdir(skill_path):
                    skill_file = os.path.join(skill_path, "skill.md")
                    if os.path.exists(skill_file):
                        self._load_skill_from_file(skill_file)

    def _load_skill_from_file(self, skill_file: str):
        """从文件加载单个技能"""
        try:
            with open(skill_file, "r", encoding="utf-8") as f:
                content = f.read()
                metadata = parse_yaml_frontmatter(content)

                skill_name = metadata.get("name", "unknown")
                self.skills_cache[skill_name] = metadata
                print(f"✅ 已加载技能: {skill_name}")
        except Exception as e:
            print(f"❌ 加载技能失败 {skill_file}: {e}")

    def get_skill(self, skill_name: str) -> Optional[dict]:
        """获取指定技能的完整内容"""
        return self.skills_cache.get(skill_name)

    def get_all_skills_metadata(self) -> list[dict]:
        """获取所有技能的元数据(不含完整内容)"""
        return [
            {
                "name": skill["name"],
                "description": skill.get("description", "无描述")
            }
            for skill in self.skills_cache.values()
        ]


# ==================== 全局技能加载器 ====================
skill_loader = None


# ==================== 技能加载工具 ====================
@tool
def load_skill(skill_name: str) -> str:
    """
    加载技能的完整内容

    Args:
        skill_name: 技能名称
    """
    global skill_loader

    if skill_loader is None:
        return "❌ 技能加载器未初始化"

    skill = skill_loader.get_skill(skill_name)

    if skill:
        return f"✅ 已加载技能: {skill_name}\n\n{skill['content']}"
    else:
        available = ", ".join([s["name"] for s in skill_loader.get_all_skills_metadata()])
        return f"❌ 未找到技能 '{skill_name}'。可用技能: {available}"


# ==================== 技能中间件 ====================
class SkillsMiddleware(AgentMiddleware):
    """将技能描述注入系统提示词的中间件"""

    tools = [load_skill]

    def __init__(self, skills_dirs: list[str]):
        """
        初始化技能中间件

        Args:
            skills_dirs: 技能目录列表
        """
        global skill_loader
        skill_loader = SkillLoader(skills_dirs)

        # 构建技能列表提示词
        skills_metadata = skill_loader.get_all_skills_metadata()
        skills_list = []
        for skill in skills_metadata:
            skills_list.append(f"- **{skill['name']}**: {skill['description']}")

        self.skills_prompt = "\n".join(skills_list) if skills_list else "暂无可用技能"

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        """将技能描述注入系统提示词"""
        skills_addendum = (
            f"\n\n## 📚 可用技能\n\n{self.skills_prompt}\n\n"
            "💡 使用 `load_skill` 工具加载技能的详细内容。"
        )

        new_content = list(request.system_message.content_blocks) + [
            {"type": "text", "text": skills_addendum}
        ]
        new_system_message = SystemMessage(content=new_content)
        modified_request = request.override(system_message=new_system_message)
        return handler(modified_request)


# ==================== 创建 Agent ====================
research_instructions = """
你是一个智能助手
"""
# 创建后端(使用当前目录)
from pathlib import Path
current_dir = Path(".").resolve()  # 获取绝对路径
print(f"工作目录: {current_dir}")
llm = ChatOpenAI(model=modelName)
backend = FilesystemBackend(root_dir=current_dir, virtual_mode=True)
skills_path = str(current_dir / ".deepagents" / "skills")
agent = create_deep_agent(
    model=llm,
    tools=[query_personal_infos],
    middleware=[SkillsMiddleware([
        skills_path
    ])],
    system_prompt=research_instructions,
    backend=backend
)

运行上面代码可以看到

在这里插入图片描述

thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "帮我查询李娜的信息并生成 HTML 文件进行保存,保存的时候名称为李娜.html"
            }
        ]
    },
    config
)

for message in result["messages"]:
    if hasattr(message, 'pretty_print'):
        message.pretty_print()
    else:
        print(f"{message.type}: {message.content}")

运行代码
在这里插入图片描述

在这里插入图片描述

八、源码层分析

大模型 modify_request _parse_skill_metadata _list_skills before_agent SkillsMiddleware create_deep_agent 用户代码 大模型 modify_request _parse_skill_metadata _list_skills before_agent SkillsMiddleware create_deep_agent 用户代码 调用 create_deep_agent(skills=["/.deepagents/skills"]) 创建 SkillsMiddleware(backend, sources) 将 SM 添加到 middleware 栈 返回 compiled_agent agent.invoke() 触发 检查 skills_metadata 是否存在 调用 _list_skills(backend, source_path) backend.ls_info(base_path) 列出目录 筛选子目录,构建 SKILL.md 路径 backend.download_files(paths) 批量下载 调用 _parse_skill_metadata(content, path, dir) 正则匹配 YAML frontmatter yaml.safe_load 解析 验证必填字段 返回 SkillMetadata 返回 skills 列表 更新 state.skills_metadata 继续执行 LLM 即将调用,触发 modify_request 从 state 获取 skills_metadata 格式化技能信息 注入到系统提示词 修改后的 request 响应(包含技能上下文)

create_deep_agent 函数

https://github.com/langchain-ai/deepagents/blob/deepagents%3D%3D0.3.6/libs/deepagents/deepagents/graph.py#L44

关键参数

def create_deep_agent(
    model: str | BaseChatModel | None = None,
    tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
    system_prompt: str | None = None,
    skills: list[str] | None = None,  # ← 技能路径列表
    backend: BackendProtocol | BackendFactory | None = None,
    # ... 其他参数
)
  • Skills 会同时注入到主 Agent 和 SubAgent
  • backend 决定文件存储方式(StateBackend / FilesystemBackend)
  • sources 是 POSIX 格式路径列表,如 [“/.deepagents/skills”]

SkillsMiddleware 初始化

https://github.com/langchain-ai/deepagents/blob/deepagents%3D%3D0.3.6/libs/deepagents/deepagents/middleware/skills.py

在这里插入图片描述

skills.py 的 init 方法:

def __init__(self, *, backend: BACKEND_TYPES, sources: list[str]) -> None:
    """初始化技能中间件
    
    Args:
        backend: 后端实例或工厂函数
        sources: 技能源路径列表 (如 ["/skills/user/", "/skills/project/"])
    """
    self._backend = backend
    self.sources = sources
    self.system_prompt_template = SKILLS_SYSTEM_PROMPT

技能加载before_agent

每次 Agent 执行前自动调用
在这里插入图片描述

Logo

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地,聚焦技能开发、插件实践与部署教程,为开发者提供可直接落地的方案、工具与交流平台,助力高效构建与落地 AI 应用

更多推荐