1. LangGraph技能开发概述
在LangGraph生态系统中,技能(Skills)是智能代理(Agent)的核心能力单元。每个技能都是一个独立的功能模块,包含执行特定任务所需的所有指令、代码和资源。与传统的函数调用不同,LangGraph技能采用声明式设计,通过SKILL.md文件描述功能和使用方式,使得非技术用户也能参与技能开发和维护。
典型的技能目录结构如下:
code复制skill-name/
├── SKILL.md # 技能主文档(含YAML元数据和操作指令)
├── scripts/ # 可执行代码(Python/Bash/JS等)
├── references/ # 参考文档(API说明、表单模板等)
└── assets/ # 静态资源(图片、数据文件等)
2. SKILL.md文件规范详解
2.1 文件结构与元数据
SKILL.md采用Markdown格式,包含YAML frontmatter和自由格式的指令内容。以下是标准模板:
markdown复制---
name: pdf-extractor
description: Extract tables and text from PDF documents
version: 1.0.2
author: dev-team@example.com
tags:
- document-processing
- data-extraction
dependencies:
- pypdf2>=3.0.0
- pandas>=2.0.0
---
# PDF提取器
## 功能说明
本技能用于从PDF文件中提取结构化数据,支持:
- 表格内容提取(转为CSV)
- 文本内容分段提取
- 元数据读取(作者、创建日期等)
## 使用示例
```bash
python scripts/extract.py --input report.pdf --output result.csv
参数说明
| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
| --input | 文件路径 | 是 | 待处理的PDF文件 |
| --output | 文件路径 | 否 | 输出文件(默认:./output.csv) |
| --pages | 字符串 | 否 | 指定页码(如"1-3,5") |
code复制
### 2.2 指令编写最佳实践
1. **任务分解**:将复杂操作拆分为原子步骤
```markdown
## 数据处理流程
1. 运行`scripts/clean.py`预处理原始数据
2. 执行`scripts/transform.py`进行格式转换
3. 使用`scripts/validate.py`检查数据质量
-
错误处理指导:
markdown复制## 常见问题 - 如果遇到编码错误:尝试指定`--encoding utf-8` - 内存不足时:添加`--chunk-size 1000`分批处理 -
多语言支持:
markdown复制## 国际化 脚本支持通过`--lang`参数指定语言: - en: 英语(默认) - zh: 中文 - ja: 日语
3. 技能开发全流程
3.1 环境准备
推荐使用conda创建隔离环境:
bash复制conda create -n langgraph-skills python=3.10
conda activate langgraph-skills
pip install langgraph deepagents pypdf2 pandas
3.2 技能初始化
使用CLI工具创建技能骨架:
bash复制langgraph skill init pdf-extractor
cd pdf-extractor
生成的目录结构包含:
SKILL.md模板文件scripts/空目录tests/测试用例目录
3.3 核心脚本开发
以PDF处理为例,创建scripts/extract.py:
python复制import argparse
from pypdf import PdfReader
import pandas as pd
def extract_tables(pdf_path, output_csv):
reader = PdfReader(pdf_path)
tables = []
for page in reader.pages:
tables.extend(page.extract_tables())
# 合并所有表格
df = pd.concat([pd.DataFrame(table[1:], columns=table[0])
for table in tables if table])
df.to_csv(output_csv, index=False)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True)
parser.add_argument("--output", default="./output.csv")
args = parser.parse_args()
extract_tables(args.input, args.output)
3.4 测试验证
创建冒烟测试脚本tests/test_extract.py:
python复制import unittest
import os
from scripts.extract import extract_tables
class TestPdfExtractor(unittest.TestCase):
@classmethod
def setUpClass(cls):
os.makedirs("test_output", exist_ok=True)
def test_table_extraction(self):
test_pdf = "test_data/sample.pdf"
output = "test_output/result.csv"
extract_tables(test_pdf, output)
self.assertTrue(os.path.exists(output))
self.assertGreater(os.path.getsize(output), 0)
if __name__ == "__main__":
unittest.main()
执行测试:
bash复制python -m unittest tests/test_extract.py
4. 技能部署与集成
4.1 本地部署
使用FilesystemBackend加载本地技能:
python复制from deepagents import create_deep_agent
from deepagents.backends.filesystem import FilesystemBackend
backend = FilesystemBackend(root_dir="/path/to/skills")
agent = create_deep_agent(
model="anthropic:claude-3-opus",
backend=backend,
skills=["/pdf-extractor"]
)
4.2 远程存储集成
从URL加载技能:
python复制from urllib.request import urlopen
from deepagents.backends import StateBackend
from deepagents.backends.utils import create_file_data
backend = StateBackend()
skill_url = "https://example.com/skills/pdf-extractor/SKILL.md"
with urlopen(skill_url) as response:
skill_content = response.read().decode('utf-8')
agent = create_deep_agent(
model="openai:gpt-4",
backend=backend,
skills={
"/skills/pdf-extractor/SKILL.md": create_file_data(skill_content)
}
)
4.3 权限控制
设置技能读写权限:
python复制from deepagents import FilesystemPermission
agent = create_deep_agent(
model="anthropic:claude-3-sonnet",
permissions=[
FilesystemPermission(
operations=["write"],
paths=["/skills/**"],
mode="interrupt" # 需要人工确认
)
]
)
5. 高级技能开发技巧
5.1 多技能组合
创建技能编排流水线:
python复制agent = create_deep_agent(
model="google_genai:gemini-pro",
skills=[
"/data-extraction/pdf-extractor",
"/data-processing/cleaner",
"/analysis/stats-analyzer"
],
workflow={
"extract": {"skill": "pdf-extractor"},
"clean": {"depends_on": "extract", "skill": "cleaner"},
"analyze": {"depends_on": "clean", "skill": "stats-analyzer"}
}
)
5.2 动态技能加载
基于上下文选择技能:
python复制SKILLS_BY_ROLE = {
"analyst": ["/skills/data-visualization", "/skills/stats-analysis"],
"engineer": ["/skills/code-review", "/skills/deployment"]
}
def create_role_agent(role):
return create_deep_agent(
model="anthropic:claude-3-haiku",
skills=SKILLS_BY_ROLE.get(role, [])
)
5.3 技能版本管理
在SKILL.md中声明版本:
yaml复制---
name: pdf-extractor
version: 1.2.0
compatibility:
min_agent_version: "0.8.0"
max_agent_version: "1.*"
---
6. 调试与优化
6.1 常见错误排查
| 错误现象 | 可能原因 | 解决方案 |
|---|---|---|
| 技能未加载 | 路径错误 | 检查backend.root_dir和skills参数 |
| 脚本执行失败 | 依赖缺失 | 在SKILL.md的dependencies中声明 |
| 权限拒绝 | 沙盒限制 | 配置sandbox_backend的权限规则 |
6.2 性能优化技巧
-
脚本预编译:对Python脚本使用
py_compilepython复制import py_compile py_compile.compile("scripts/extract.py") -
缓存机制:对耗时操作添加缓存
python复制from functools import lru_cache @lru_cache(maxsize=100) def process_data(data): # 复杂计算... -
分批处理:大数据集分块处理
python复制chunk_size = 1000 for i in range(0, len(data), chunk_size): process_batch(data[i:i+chunk_size])
7. 技能商店与分发
7.1 打包技能
创建技能包:
bash复制langgraph skill pack pdf-extractor --output dist/pdf-extractor-1.0.0.lgs
7.2 发布到商店
通过CLI发布:
bash复制langgraph store login
langgraph store publish dist/pdf-extractor-1.0.0.lgs
7.3 从商店安装
用户安装命令:
bash复制langgraph skill install official/pdf-extractor
8. 实战案例:构建简历解析技能
8.1 技能设计
markdown复制---
name: resume-parser
description: Extract structured data from resumes in PDF/DOCX format
inputs:
- file: 简历文件
outputs:
- json: 结构化数据
---
### 解析流程
1. 文件类型检测(PDF/DOCX)
2. 文本内容提取
3. 实体识别(姓名、联系方式等)
4. 工作经历分段
5. 技能标签提取
8.2 核心实现
scripts/parse_resume.py关键部分:
python复制def parse_pdf(file):
from pdfminer.high_level import extract_text
text = extract_text(file)
return parse_common(text)
def parse_docx(file):
from docx import Document
doc = Document(file)
text = "\n".join(p.text for p in doc.paragraphs)
return parse_common(text)
def parse_common(text):
# 实现通用解析逻辑
return {
"name": extract_name(text),
"contact": extract_contact(text),
"experience": extract_experience(text)
}
8.3 集成测试
模拟端到端测试:
python复制def test_resume_parser():
agent = create_test_agent("/skills/resume-parser")
result = agent.invoke({
"action": "parse_resume",
"file": "test_data/john_doe.pdf"
})
assert "name" in result
assert "experience" in result
9. 技能演进与维护
9.1 版本升级策略
-
语义化版本:
- MAJOR:不兼容的API修改
- MINOR:向下兼容的功能新增
- PATCH:向下兼容的问题修正
-
弃用周期:
markdown复制## 版本说明 - 1.1.0:新增JSON输出格式(推荐) - 1.0.0:旧版XML格式(已弃用,将在2.0.0移除)
9.2 用户反馈收集
在SKILL.md中添加反馈渠道:
markdown复制## 问题报告
遇到问题请提交到:
[Issues](https://github.com/your-repo/issues)
## 功能建议
欢迎通过邮件提出建议:
skills-feedback@example.com
10. 安全最佳实践
10.1 输入验证
所有脚本应包含输入检查:
python复制def safe_path(input_path):
"""防止路径遍历攻击"""
if '../' in input_path:
raise ValueError("Invalid path")
return os.path.abspath(input_path)
10.2 沙盒执行
危险操作应在沙盒中运行:
python复制from deepagents.backends.langsmith import LangSmithSandbox
sandbox = LangSmithSandbox()
result = sandbox.run("python risky_script.py", timeout=30)
10.3 权限最小化
python复制agent = create_deep_agent(
permissions=[
FilesystemPermission(
paths=["/tmp/**"],
operations=["read", "write"],
mode="allow"
)
]
)
