探讨RAG技术、AI代理及AI伦理挑战
导读:如果说大模型(LLM)是 AI Agent 的“大脑”,那么 Skills(技能/工具)就是它的“双手”。没有 Skills,大模型只是一个博学的“空谈家”;有了 Skills,它才能真正连接业务,成为干活的“实干家”。

本文将带你从零构建一套生产级可用的 Skills 实战架构,不讲空话,直接上代码和模板。
在早期的 Prompt Engineering 中,我们试图通过“角色扮演”让模型具备某种能力。但在 Enterprise(企业级)落地中,这远远不够。
真正的 Skills(技能)必须包含三个核心要素,缺一不可:
在编写代码之前,我们需要理解 Agent 调用 Skills 的标准生命周期(Observe-Think-Act):
假设我们要开发一个“企业数据查询技能” (EnterpriseDataSkill),用于查询公司内部的销售数据。我们将使用 Python 和 Pydantic 来实现标准化的 Schema 定义。
使用 Pydantic 可以自动生成标准 JSON Schema,这是各大模型(OpenAI/Claude/DeepSeek)通用的“接口语言”。
from pydantic import BaseModel, Field
from typing import Optional, Literal
# 1. 定义参数结构
class SalesQueryArgs(BaseModel):
region: str = Field(..., descriptinotallow="查询的销售区域,例如:'华东', '北美'")
quarter: str = Field(..., descriptinotallow="查询的季度,格式如 '2024-Q1'")
product_line: Optional[str] = Field(None, descriptinotallow="可选,特定的产品线名称")
# 2. 定义 Skill 元数据
SKILL_METADATA = {
"name": "query_sales_data",
"description": "查询公司特定区域和季度的销售业绩数据。当用户询问'XX地区上个季度卖了多少'时使用此工具。",
"parameters": SalesQueryArgs.model_json_schema()
}
这是实际的业务逻辑代码。在生产环境中,这里通常涉及数据库连接或 API 调用。
import json
def query_sales_data(region: str, quarter: str, product_line: str = None):
"""
模拟数据库查询逻辑
"""
print(f"DEBUG: 正在查询数据库... Reginotallow={region}, Quarter={quarter}")
# 模拟返回结果
mock_db = {
"华东": {"2024-Q1": 1500000},
"北美": {"2024-Q1": 2000000}
}
amount = mock_db.get(region, {}).get(quarter, 0)
result = {
"status": "success",
"data": {
"region": region,
"quarter": quarter,
"amount": amount,
"currency": "CNY"
}
}
# 注意:Skills 的返回值通常必须是 String 类型,以便喂回给 LLM
return json.dumps(result, ensure_ascii=False)
有了“骨架”和“肌肉”,现在我们需要给 Agent 注入“神经信号”。
我们将使用 Anthropic SDK 实现一个标准的 ReAct (Reason + Act) 闭环。这里的核心在于处理 tool_use (模型请求) 和 tool_result (执行反馈) 的“乒乓球”交互。
代码实战:
import anthropic
from typing import List, Dict, Any
# 假设 client 已初始化
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
def run_sales_agent(user_query: str):
print(f"👤 User: {user_query}")
# 1. 动态生成工具定义 (直接复用第一步的 Pydantic Schema)
# 这是自动化关键:代码改了,Schema 自动变,不用手动维护 JSON
tools_schema = [{
"name": "query_sales_data",
"description": SKILL_METADATA["description"],
"input_schema": SalesQueryArgs.model_json_schema()
}]
# 2. 初始化对话上下文
messages = [{"role": "user", "content": user_query}]
# --- Round 1: 发球 (Claude 思考) ---
response = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
tools=tools_schema,
messages=messages
)
# 3. 拦截:判断模型是否想“拿”工具
if response.stop_reason == "tool_use":
# 必须将 Claude 的"思考过程"完整加入历史,否则上下文会断裂
messages.append({"role": "assistant", "content": response.content})
# 提取核心信息
tool_block = next(b for b in response.content if b.type == "tool_use")
tool_id = tool_block.id # 身份证:用于后续匹配结果
call_args = tool_block.input # 参数包:{'region': '华东', 'quarter': '2024-Q1'}
print(f"🤖 Agent: 正在调用工具 -> query_sales_data({call_args})")
# --- Round 2: 接球 (本地执行) ---
# 这里执行我们在第二步写好的业务函数
try:
# **关键点**:使用 **kwargs 自动解包参数,直接传给函数
execution_result = query_sales_data(**call_args)
except Exception as e:
execution_result = f"Error: {str(e)}"
print(f"✅ System: 执行成功,数据已获取。")
# --- Round 3: 扣杀 (反馈结果) ---
# 构造符合 Anthropic 协议的反馈块
tool_result_message = {
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_id, # 必须与请求 ID 严格对应
"content": execution_result
}]
}
messages.append(tool_result_message)
# 让 Claude 根据数据生成最终的自然语言回答
final_response = client.messages.create(
model="claude-3-5-sonnet-20240620",
max_tokens=1024,
tools=tools_schema, # 保持工具定义一致
messages=messages
)
return final_response.content[0].text
# 如果模型不需要工具,直接返回文本
return response.content[0].text
# === 🚀 运行效果 ===
# output = run_sales_agent("帮我查一下华东区2024年第一季度的业绩")
# print(f"💬 Final Answer: {output}")
核心逻辑拆解:
input_schema: SalesQueryArgs.model_json_schema()。这意味着你以后只要修改 Pydantic 类(比如增加一个字段),工具定义会自动更新,无需手动改 JSON,这是工程化的关键细节。tool_use_id 是唯一的。你必须把它原封不动地塞回tool_result 中,这样模型才能知道“这个结果是对应刚才那次查询的”。很多开发者代码写得很好,但模型就是不调,或者参数乱填。核心原因在于 Context Engineering(上下文工程)没做好。
以下是一套经过实战验证的 System Prompt 注入模板:
📝 通用 Skill 注入模板
# Role
你是一个配备了专业工具的 AI 业务助手。你的目标是协助用户查询数据和解决问题。
# Tools Capability
你拥有以下工具(Skills)的访问权限:
## 1. query_sales_data
- **功能描述**: {SKILL_DESCRIPTION}
- **参数要求**: {JSON_SCHEMA}
- **触发时机**: 当且仅当用户明确询问有关“销售额”、“业绩”或“营收”等定量数据时调用。对于一般性闲聊(如“你好”),请勿调用此工具。
# Constraints (关键约束)
1. **参数推断**: 如果用户没有提供必要的参数(如只说了“查一下销售额”,没说哪个区),你必须先追问用户,而不是编造参数。
2. **格式严格**: 调用工具时,必须输出严格的 JSON 格式。
3. **事实导向**: 永远依据工具返回的数据回答,不要使用你预训练知识中的过时数据。
💡 技巧点拨:
在将 Skills 部署到生产环境时,这三个坑一定要避开:
a.问题: 如果你挂载了 50 个工具,光是工具定义的 JSON Schema 就会把 Token 撑爆。
b.解法: 使用 RAG 技术动态加载 Skills。先根据用户 Query 检索出最相关的 Top-5 工具,再将这 5 个工具的定义注入 Prompt,示例代码如下:
import json
import numpy as np
from typing import List, Dict
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
# 1. 模拟一个包含大量 Skills 的工具库
# 在真实场景中,这里通常是一个数据库或配置文件
ALL_SKILLS_REPOSITORY = [
{
"name": "query_sales_data",
"description": "查询公司特定区域和季度的销售业绩数据、营收报表。",
"schema": {"type": "function", "function": {"name": "query_sales_data", "parameters": {...}}}
},
{
"name": "check_weather",
"description": "查询此时此刻的全球各城市天气情况、温度、降雨概率。",
"schema": {"type": "function", "function": {"name": "check_weather", "parameters": {...}}}
},
{
"name": "send_feishu_message",
"description": "向飞书/Lark群组或个人发送通知消息、提醒。",
"schema": {"type": "function", "function": {"name": "send_feishu_message", "parameters": {...}}}
},
{
"name": "search_internal_wiki",
"description": "搜索公司内部Wiki知识库,查找技术文档、HR政策等。",
"schema": {"type": "function", "function": {"name": "search_internal_wiki", "parameters": {...}}}
},
# ... 假设这里还有 50 个工具 ...
]
class SkillRetriever:
def __init__(self, skills: List[Dict]):
self.skills = skills
# 加载一个轻量级的 Embedding 模型
print("正在加载 Embedding 模型 (首次运行可能较慢)...")
self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
# 预计算所有 Skill Description 的向量并缓存
descriptions = [skill["description"] for skill in skills]
self.skill_embeddings = self.encoder.encode(descriptions)
print(f"✅ 已索引 {len(skills)} 个 Skills。")
def retrieve(self, user_query: str, top_k: int = 2) -> List[Dict]:
"""
根据用户输入,动态检索最相关的 Top-K 个工具
"""
# 1. 将用户 Query 向量化
query_embedding = self.encoder.encode([user_query])
# 2. 计算余弦相似度 (Cosine Similarity)
similarities = cosine_similarity(query_embedding, self.skill_embeddings)[0]
# 3. 获取 Top-K 索引
# argsort 返回的是从小到大的索引,所以取最后 k 个并反转
top_k_indices = np.argsort(similarities)[-top_k:][::-1]
# 4. 返回对应的 Skill 对象
selected_skills = []
print(f"\n🔍 用户意图: '{user_query}'")
print(f"🎯 命中工具 (Top {top_k}):")
for idx in top_k_indices:
score = similarities[idx]
skill = self.skills[idx]
print(f" - {skill['name']} (相似度: {score:.4f})")
selected_skills.append(skill)
return selected_skills
# --- 测试运行 ---
# 初始化检索器
retriever = SkillRetriever(ALL_SKILLS_REPOSITORY)
# 场景 A: 用户想查数据
relevant_skills_a = retriever.retrieve("帮我看看上个季度华东区的业绩怎么样", top_k=2)
# 场景 B: 用户想发通知
relevant_skills_b = retriever.retrieve("给项目组发个消息,说服务器修好了", top_k=2)
a.问题: Skill 代码报错了(比如数据库超时),直接把 Python Traceback 丢给 LLM?
b.解法: 永远要在 Skill 内部捕获异常,并返回一段“人类可读”的错误描述。
c.代码示例:
try:
# 业务逻辑
except TimeoutError:
return "系统提示:数据源连接超时,请建议用户稍后重试。"
a.问题: 允许 LLM 执行 delete_user 或 drop_table?
b.解法: 读写分离。对于高危操作(增删改),必须在 Skill 执行前增加一步 "Human-in-the-loop" (人工确认) 环节。
开发 AI Agent 的过程,本质上是将人类的“业务 SOP”转化为“代码逻辑(Skills)”和“自然语言逻辑(Prompt)”的过程。
掌握了 Schema 定义 + 业务代码 + 指令模板这套组合拳,你就掌握了构建强大 Agent 的钥匙。
好了,这就是我今天想分享的内容。
关注公众号
立刻获取最新消息及人工咨询