LangGraph 7. 技能 Skills（附完整免费源代码）

# 基于提示词工程使用
LangChain及
LangGraph开发Skill的
完整教程 1
. 项目架构设计 1
.1 三层架构设计模式 基于跨平台Agent Skill开发的最佳实践，我们采用三层架构模式来构建Skill系统[ref_4]： python from abc import ABC, abstractmethod from typing import Dict, Any, List import jinja2 from
langchain_core
.prompts import ChatPromptTemplate from
langgraph
.graph import StateGraph, END from pydantic import BaseModel # 抽象Skill接口定义 class BaseSkill
(ABC
): @abstractmethod def get_name
(self
) -> str: pass @abstractmethod def get_description
(self
) -> str: pass @abstractmethod def execute
(self, state: Dict[str, Any]
) -> Dict[str, Any]: pass # 模型适配器接口 class ModelAdapter
(ABC
): @abstractmethod def format_prompt
(self, skill: BaseSkill, input_data: Dict
) -> str: pass @abstractmethod def parse_response
(self, response: str
) -> Dict[str, Any]: pass 1
.2 状态管理设计
LangGraph的核心优势在于状态管理，我们设计统一的状态模型[ref_2]： python from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator class
SkillState
(TypedDict
): # 输入相关 user_input: str current_skill: str skill_parameters: Dict[str, Any] # 处理过程 intermediate_steps: Annotated[List[Dict], operator
.add] prompt_history: Annotated[List[str], operator
.add] # 输出相关 final_output: str execution_status: str # “success”, “error”, “pending” confidence_score: float 2
. 提示词工程优化实现 2
.1 三段式提示词结构 采用标准化的三段式提示词结构，确保跨模型兼容性[ref_4]： python class PromptOptimizer: def __init__
(self
): self
.template_loader = jinja2
.Environment
( loader=jinja2
.FileSystemLoader
(“templates”
), autoescape=jinja2
.select_autoescape
(
)
) def create_three_part_prompt
(self, skill_name: str, context: Dict
) -> str: “””创建三段式提示词：角色定义、任务描述、输出格式””” template = self
.template_loader
.get_template
(“three_part_prompt
.j2″
) prompt_structure = return template
.render
(prompt_structure
) def _get_role_definition
(self, skill_name: str
) -> str: roles = { “calculator”: “你是一个专业的数学计算助手，擅长进行各种数学运算。”, “searcher”: “你是一个信息检索专家，能够准确理解用户查询意图并获取相关信息。”, “analyzer”: “你是一个数据分析师，能够对提供的数据进行深入分析和洞察。” } return roles
.get
(skill_name, “你是一个专业的AI助手。”
) def _get_output_format
(self, skill_name: str
) -> str: formats = { “calculator”: “请以JSON格式返回计算结果：{‘result’: 数值, ‘steps’: 计算步骤}”, “searcher”: “请返回搜索结果：{‘query’: 搜索词, ‘results’: 相关信息列表}”, “analyzer”: “请返回分析报告：{‘insights’: 关键发现, ‘recommendations’: 建议}” } return formats
.get
(skill_name, “请以清晰的结构返回结果。”
) 2
.2 动态提示词生成 基于
LangChain的提示词模板系统实现动态提示词生成[ref_6]： python from
langchain
.prompts import
( ChatPromptTemplate, HumanMessagePromptTemplate, SystemMessagePromptTemplate
) def create_dynamic_prompt_template
(skill_type: str
) -> ChatPromptTemplate: “””创建动态提示词模板””” system_template = SystemMessagePromptTemplate
.from_template
( “你是一个{skill_type}专家。请根据用户输入提供专业的响应。 ” “输出格式要求：{output_format} ” “安全注意事项：{safety_notes}”
) human_template = HumanMessagePromptTemplate
.from_template
( “用户输入：{user_input} ” “上下文信息：{context} ” “请执行{skill_type}操作。”
) return ChatPromptTemplate
.from_messages
([system_template, human_template]
) # 示例：计算器Skill的提示词模板 calculator_prompt = create_dynamic_prompt_template
( skill_type=”数学计算”, output_format=”{‘result’: 数值结果, ‘steps’: 详细计算步骤, ‘confidence’: 置信度}”, safety_notes=”确保计算准确，避免除零错误等数学问题”
) 3
. 具体Skill实现示例 3
.1 计算器Skill
完整实现 python class CalculatorSkill
(BaseSkill
): def __init__
(self, model_adapter: ModelAdapter
): self
.name = “calculator” self
.description = “执行数学计算任务” self
.model_adapter = model_adapter self
.prompt_optimizer = PromptOptimizer
(
) def get_name
(self
) -> str: return self
.name def get_description
(self
) -> str: return self
.description def execute
(self, state:
SkillState
) ->
SkillState: try: # 生成优化后的提示词 prompt = self
.prompt_optimizer
.create_three_part_prompt
( “calculator”, {“expression”: state[“user_input”]}
) # 使用模型适配器处理 formatted_input = self
.model_adapter
.format_prompt
(self, {“prompt”: prompt}
) # 模拟模型调用（实际项目中替换为真实模型调用） response = self
._call_math_model
(formatted_input
) # 解析响应 result = self
.model_adapter
.parse_response
(response
) # 更新状态 state
.update
({ “final_output”: result, “execution_status”: “success”, “confidence_score”: 0
.95, “intermediate_steps”: Agent 智能体 state[“intermediate_steps”] + [{ “step”: “calculation”, “input”: state[“user_input”], “output”: result }] }
) except Exception as e: state
.update
({ “final_output”: f”计算错误: {str
(e
)}”, “execution_status”: “error”, “confidence_score”: 0
.0 }
) return state def _call_math_model
(self, input_data: Dict
) -> str: “””模拟数学计算模型调用””” # 实际项目中这里会集成真实的LLM调用 expression = input_data
.get
(“expression”, “”
) try: # 简单的表达式计算（生产环境使用更安全的计算方式） result = eval
(expression
) return f'{{“result”: {result}, “steps”: “计算表达式: {expression}”, “confidence”: 0
.95}}’ except Exception as e: return f'{{“error”: “{str
(e
)}”, “confidence”: 0
.0}}’ 3
.2 搜索Skill实现 python class SearchSkill
(BaseSkill
): def __init__
(self, model_adapter: ModelAdapter, search_tool
): self
.name = “searcher” self
.description = “执行联网搜索任务” self
.model_adapter = model_adapter self
.search_tool = search_tool self
.prompt_optimizer = PromptOptimizer
(
) def execute
(self, state:
SkillState
) ->
SkillState: # 生成搜索查询优化提示词 query_optimization_prompt = self
.prompt_optimizer
.create_three_part_prompt
( “searcher”, {“original_query”: state[“user_input”]}
) # 优化搜索查询 optimized_query = self
._optimize_search_query
(query_optimization_prompt
) # 执行搜索 search_results = self
.search_tool
.search
(optimized_query
) # 结果处理和分析 analysis_prompt = self
._create_analysis_prompt
(state[“user_input”], search_results
) analyzed_results = self
._analyze_search_results
(analysis_prompt
) state
.update
({ “final_output”: analyzed_results, “execution_status”: “success”, “confidence_score”: 0
.85, “intermediate_steps”: state[“intermediate_steps”] + [{ “step”: “search_execution”, “optimized_query”: optimized_query, “raw_results”: search_results }] }
) return state def _optimize_search_query
(self, prompt: str
) -> str: “””优化搜索查询词””” # 实际项目中调用LLM优化查询 return self
.user_input # 简化实现 4
.
LangGraph工作流编排 4
.1 Skill路由和工作流图 python class SkillRouter: def __init__
(self
): self
.
skills = {} self
.graph = StateGraph
(
SkillState
) def register_skill
(self, skill: BaseSkill
): “””注册Skill到路由系统””” self
.
skills[skill
.get_name
(
)] = skill # 动态添加节点到图 self
.graph
.add_node
(skill
.get_name
(
), skill
.execute
) def build_workflow
(self
): “””构建工作流图””” # 定义路由决策函数 def route_
skills
(state:
SkillState
) -> str: user_input = state[“user_input”]
.lower
(
) # 基于关键词的路由逻辑 if any
(keyword in user_input for keyword in [“计算”, “算一下”, “+”, “-“, “*”, “/”]
): return “calculator” elif any
(keyword in user_input for keyword in [“搜索”, “查找”, “查询”]
): return “searcher” elif any
(keyword in user_input for keyword in [“分析”, “统计”, “报告”]
): return “analyzer” else: return “general_agent” # 设置图结构 self
.graph
.add_conditional_edges
( “start”, route_
skills, { “calculator”: “calculator”, “searcher”: “searcher”, “analyzer”: “analyzer”, “general_agent”: “general_agent” }
) # 所有
技能节点都连接到结束 for skill_name in self
.
skills
.keys
(
): self
.graph
.add_edge
(skill_name, END
) self
.graph
.set_entry_point
(“start”
) return self
.graph
.compile
(
) 4
.2
完整工作流执行示例 python def create_complete_skill_agent
(
): “””创建
完整的Skill Agent系统””” # 初始化组件 model_adapter = OpenAIModelAdapter
(
) # 实际实现模型适配器 search_tool = SerperSearchTool
(
) # 实际实现搜索工具 # 创建Skill实例 calculator_skill = CalculatorSkill
(model_adapter
) search_skill = SearchSkill
(model_adapter, search_tool
) # 构建路由和工作流 router = SkillRouter
(
) router
.register_skill
(calculator_skill
) router
.register_skill
(search_skill
) # 编译工作流图 workflow = router
.build_workflow
(
) return workflow # 使用示例 def run_skill_agent_example
(
): agent = create_complete_skill_agent
(
) # 初始化状态 initial_state =
SkillState
( user_input=”计算一下 125 * 48 等于多少？”, current_skill=””, skill_parameters={}, intermediate_steps=[], prompt_history=[], final_output=””, execution_status=”pending”, confidence_score=0
.0
) # 执行工作流 result = agent
.invoke
(initial_state
) print
(“最终结果:”, result[“final_output”]
) print
(“执行状态:”, result[“execution_status”]
) print
(“置信度:”, result[“confidence_score”]
) return result 5
. 高级特性与最佳实践 5
.1 渐进式
技能披露 基于
LangChain的渐进式披露机制，实现按需加载
技能[