大模型开发手记（十三）：langchain skills（下）：构建skills架构agent实战

目录

• 前言
• 一、整体架构预览
• 二、实战
• • 2.1 第一步：定义Skill文件
  • • 酒店预订Skill
    • 景点推荐Skill
  • 2.2 第二步：编写Skill加载工具
  • 2.3 第三步：构建Skill中间件
  • 2.4 第四步：创建agent.py：
  • 第四步：运行与验证
• 三、扩展思路

前言

在上一篇博客中，我们深入理解了Agent Skill的核心概念——它是一种通过渐进式披露来扩展Agent专业能力的模块化方式。理论说完了，今天我们就来真刀真枪地实现一个。

我们的目标：构建一个旅行助手Agent，它拥有两个独立的Skill：

• 酒店预订Skill：包含酒店预订的业务规则、政策、操作流程。
• 景点推荐Skill：包含热门景点的介绍、推荐逻辑、注意事项。

Agent启动时只知道这两个Skill的名字和一句话描述，只有当你问及具体问题时，它才会动态加载对应Skill的详细内容。

一、整体架构预览

travel–agent/
├── skills/ # 存放所有Skill
│ ├── hotel_booking/ # 酒店预订Skill目录
│ │ ├── skill.json # 元数据（name, description）
│ │ └── content.md # 详细内容（业务规则、示例等）
│ └── attraction/ # 景点推荐Skill目录
│ ├── skill.json
│ └── content.md
├── agent.py # 主程序：构建Agent、加载Skill
└── skill_loader.py # 工具函数：从文件系统加载Skill

核心流程：

二、实战

2.1 第一步：定义Skill文件

每个Skill都是一个独立的目录，元数据和内容分离。未来新增Skill只需新建一个子目录并放入对应的skill.json和content.md即可。

酒店预订Skill

景点推荐Skill

2.2 第二步：编写Skill加载工具

创建skill_loader.py，负责扫描目录、加载元数据和按需加载内容：

import json
from pathlib import Path
from typing import Dict, List, Optional

from langchain.tools import tool

# Skill根目录
SKILLS_ROOT = Path(__file__).parent / "skills"

class SkillLoader:
"""负责加载Skill元数据和内容"""

def __init__(self, skills_root: Path):
self.skills_root = skills_root
self._metadata_cache: Optional[List[Dict[str, str]]] = None

def load_all_metadata(self) –> List[Dict[str, str]]:
"""扫描所有Skill目录，返回元数据列表"""
if self._metadata_cache is not None:
return self._metadata_cache

metadata_list = []
for skill_dir in self.skills_root.iterdir():
if not skill_dir.is_dir():
continue
json_path = skill_dir / "skill.json"
if not json_path.exists():
continue
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
metadata_list.append({
"name": data["name"],
"description": data["description"]
})
self._metadata_cache = metadata_list
return metadata_list

def load_full_content(self, skill_name: str) –> Optional[str]:
"""根据技能名称加载完整的content.md内容"""
for skill_dir in self.skills_root.iterdir():
if not skill_dir.is_dir():
continue
json_path = skill_dir / "skill.json"
if not json_path.exists():
continue
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
if data["name"] == skill_name:
content_path = skill_dir / "content.md"
if content_path.exists():
with open(content_path, "r", encoding="utf-8") as cf:
return f"# 已加载技能：{skill_name}\\n\\n{cf.read()}"
else:
return f"技能 {skill_name} 缺少 content.md 文件"
return None

# 全局加载器实例
loader = SkillLoader(SKILLS_ROOT)

@tool
def load_skill(skill_name: str) –> str:
"""按需加载技能的详细内容。

当你需要某个技能的具体信息（如业务规则、操作流程、详细知识）时，
调用此工具加载对应技能。

Args:
skill_name: 技能名称，可选值由系统提示词中的可用技能列表提供。
"""
content = loader.load_full_content(skill_name)
if content is None:
available = ", ".join(s["name"] for s in loader.load_all_metadata())
return f"未找到名为 {skill_name} 的技能。可用技能：{available}"
return content

这个类做了几件事：

• load_all_metadata：扫描skills/下的所有子目录，读取skill.json，提取name和description。
• load_full_content：根据技能名称找到对应的目录，读取其中的content.md文件，返回完整的Markdown内容。
• 通过@tool装饰器将load_skill暴露给Agent使用。

2.3 第三步：构建Skill中间件

中间件负责在模型调用前将技能元数据注入系统提示词。通过模型上下文的方式（wrap_model_call中间件），在模型调用前将所有skills的元数据（名称+描述），注入到系统提示词中，让agent清晰地知道自己有哪些技能。

import uuid
from typing import Callable

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
from langchain.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver

from skill_loader import loader, load_skill

class SkillMiddleware(AgentMiddleware):
"""将技能列表注入系统提示词的中间件"""

# 注册工具，让Agent可以调用load_skill
tools = [load_skill]

def __init__(self):
# 启动时加载所有技能的元数据
self.skills_metadata = loader.load_all_metadata()
self.skills_prompt = self._build_skills_prompt()

def _build_skills_prompt(self) –> str:
"""生成技能列表的描述文本"""
if not self.skills_metadata:
return "当前没有可用技能。"

lines = ["## 可用技能列表\\n"]
for skill in self.skills_metadata:
lines.append(f"- **{skill['name']}**：{skill['description']}")
lines.append("\\n当你需要某个技能的详细内容时，请调用 `load_skill` 工具加载。")
return "\\n".join(lines)

def wrap_model_call(
self,
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) –> ModelResponse:
"""在调用模型前，将技能描述追加到系统消息中"""
# 构建追加的内容
addendum = f"\\n\\n{self.skills_prompt}"

# 获取原有系统消息的内容块，并追加新内容
original_blocks = request.system_message.content_blocks
new_blocks = original_blocks + [{"type": "text", "text": addendum}]
new_system_message = SystemMessage(content=new_blocks)

# 用新的系统消息覆盖原请求
modified_request = request.override(system_message=new_system_message)
return handler(modified_request)

2.4 第四步：创建agent.py：

创建agent，配置中间件，在invoke或者stream 启动时会将 所有skills的元数据（名称＋描述）注入到系统提示词中，然后agent运行中会调用工具将skills核心内容加载到上下文中，发送给大模型

def main():
# 初始化模型（请替换为你的API Key或使用其他模型）
model = ChatOpenAI(model="gpt-4o-mini")

# 创建Agent，使用我们的中间件
agent = create_agent(
model,
system_prompt="你是一个友好的旅行助手，帮助用户解决酒店预订和景点推荐等问题。",
middleware=[SkillMiddleware()],
checkpointer=InMemorySaver(),
)

# 对话线程ID（用于维护状态）
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

# 测试1：询问酒店预订
print("=== 测试：酒店预订 ===")
result = agent.invoke(
{"messages": [{"role": "user", "content": "我想订一间海景大床房，住两晚，多少钱？"}]},
config
)
for msg in result["messages"]:
if hasattr(msg, "pretty_print"):
msg.pretty_print()
else:
print(f"{msg.type}: {msg.content}")

print("\\n" + "="*50 + "\\n")

# 测试2：询问景点推荐
print("=== 测试：景点推荐 ===")
result = agent.invoke(
{"messages": [{"role": "user", "content": "带孩子去哪里玩比较好？"}]},
config
)
for msg in result["messages"]:
if hasattr(msg, "pretty_print"):
msg.pretty_print()
else:
print(f"{msg.type}: {msg.content}")

if __name__ == "__main__":
main()

第四步：运行与验证

三、扩展思路

这个基础实现已经足够演示核心机制，但在实际项目中你可能会需要以下增强：