第 6 章:Agent 插件与 Skills
第 6 章:Agent 插件与 Skills
关于本课
本课程的视频是某个时间点的快照。Strands 一直在活跃开发,所以本页里的代码反映的是最新写法,但视频中讲的概念依然成立。拿不准的时候,以代码为准。
本课代码:samples/06-plugins-skills
插件
插件是 Strands 的扩展机制。它让你把 Hook、Tool 和自定义逻辑打包成可复用的组件,直接装到任意 Agent 上。Hook 只是单个行为,插件则把若干基础能力捆成一个整体,方便共享和复用。
Skills 是插件的一种,提供按需加载的工作流指令。Strands 还自带别的插件,你也可以自己写,用在 Memory、编排、可观测性、治理或自定义工作流上。
Skills
Agent 越做越大,system prompt 就会被无关指令撑得越来越肿。假设你的 Agent 支持二十种工作流,你肯定不希望这二十种指令在每次请求时都占着 token。
Skills 用**渐进式披露(progressive disclosure)**解决这个问题:启动时 Agent 只看到 skill 的名字,需要时才加载完整指令。每个 skill 就是子文件夹里的一个 markdown 文件,装着针对某个具体任务的指引。如果某个 skill 跟当前请求无关,它的指令就永远进不了上下文窗口。这是上下文工程的一个例子。
from strands import Agent, AgentSkills, tool
# Tools for customer service operations
@tool
def lookup_customer(customer_id: str) -> str:
"""Look up a customer by their ID."""
# ... database/API call ...
@tool
def get_order_history(customer_id: str) -> str:
"""Get order history for a customer."""
# ...
@tool
def process_refund(order_id: str, amount: float) -> str:
"""Process a refund for an order."""
# ...
# Skills plugin discovers and registers all skills in the directory
skills_plugin = AgentSkills(skills=["./skills"])
agent = Agent(
tools=[lookup_customer, get_order_history, process_refund],
plugins=[skills_plugin],
system_prompt="""You are a customer service agent. When a customer needs help,
activate the appropriate skill for step-by-step guidance."""
)📂 customer_service.py · skills/
示例里的 skills/ 目录有四个 skill:account-troubleshooting、order-tracking、pdf-processing 和 refund-processing。每一个都是一个文件夹,里面有 SKILL.md,写明什么时候该用它、要按哪些步骤走。这个客服 Agent 是后面整门课的贯穿示例。
Goal Loop
第 1 课说过,好的 Harness 会去验证 Agent 的动作是不是真的奏效。GoalLoop 就是干这件事的插件。你来定义什么叫“完成”,把它挂到 Agent 上,重试循环由它管:
- Agent 处理 prompt 并给出回复。
- GoalLoop 用你的校验逻辑检查结果。
- 通过,循环以“satisfied”结束。
- 不通过且还有预算,GoalLoop 把反馈作为新的 user 消息注入,再次调用 Agent。
- 尝试次数或超时用完,循环停止,不再重试。
校验方式可以是一句自然语言的目标(交给第二个 Model 判定),也可以是一个普通函数。函数更有意思,因为它能跑真实的检查。下面这个例子里,测试套件不过,coding agent 就不算完;测试失败时,输出会直接作为下一条指令喂回给 Agent:
import asyncio
from strands import Agent
from strands.vended_plugins.goal import GoalLoop
from strands.vended_tools import file_editor, shell
async def tests_pass(response, agent):
proc = await asyncio.create_subprocess_exec(
"pytest", "--tb=short",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await proc.communicate()
if proc.returncode == 0:
return True
output = (stdout.decode() + stderr.decode())[-4000:]
return {
"passed": False,
"feedback": f"pytest exited {proc.returncode}.\n{output}",
}
agent = Agent(
tools=[file_editor, shell],
plugins=[GoalLoop(goal=tests_pass, max_attempts=10)],
)
agent("Fix the failing tests in tests/test_orders.py")每次重试,Agent 都会看到上一次 pytest 的原始输出。测试全绿就停,10 次用完也停,看谁先到。生产环境里一定要设 max_attempts 或 timeout,别让循环跑飞了。