LangGraph Skill 入门指南：从零构建你的第一个智能工作流

1次阅读

共计 2482 个字符，预计需要花费 7 分钟才能阅读完成。

LangGraph Skill 是一个用于构建智能工作流的工具，特别适合处理需要多个步骤协作完成的复杂任务。比如，你想开发一个天气查询功能，可能需要先验证用户输入，然后调用天气 API，最后格式化返回结果。LangGraph Skill 能帮你把这些步骤组织成一个清晰的工作流。

在开始之前，我们需要了解几个关键概念：

节点 (Node): 工作流中的一个独立处理单元，比如验证输入或调用 API
边 (Edge): 定义节点之间的流向关系，决定工作流的执行顺序
工作流 (Workflow): 由节点和边组成的有向图，表示完整的处理流程

让我们通过一个实际的天气查询案例来学习如何使用 LangGraph Skill。

首先，我们需要创建一个节点来处理天气 API 查询。这个节点会接收城市名称作为输入，然后返回天气数据。

from langgraph.skill import Node

class WeatherQueryNode(Node):
    def __init__(self):
        super().__init__(name="weather_query")

    async def execute(self, inputs):
        """
        执行天气查询
        :param inputs: 包含城市名称的字典 {'city': '北京'}
        :return: 天气数据字典
        """city = inputs.get('city')
        # 这里应该是实际的 API 调用代码
        # 为了示例，我们返回模拟数据
        return {
            'city': city,
            'weather': '晴天',
            'temperature': '25°C'
        }

在调用 API 前，我们应该先验证用户输入是否有效。

class InputValidationNode(Node):
    def __init__(self):
        super().__init__(name="input_validation")

    async def execute(self, inputs):
        """
        验证输入的城市名称
        :param inputs: 包含城市名称的字典 {'city': '北京'}
        :return: 验证后的城市名称或错误信息
        """city = inputs.get('city','').strip()
        if not city:
            return {'error': '城市名称不能为空'}
        if len(city) > 20:
            return {'error': '城市名称过长'}
        return {'city': city}

工作流中可能会出现各种错误，我们需要妥善处理它们。

class ErrorHandlerNode(Node):
    def __init__(self):
        super().__init__(name="error_handler")

    async def execute(self, inputs):
        """
        处理工作流中的错误
        :param inputs: 包含错误信息的字典
        :return: 格式化后的错误信息
        """error = inputs.get('error')
        return {
            'status': 'error',
            'message': f'处理请求时出错: {error}'
        }

现在，我们把所有节点组合成一个完整的工作流。

from langgraph.skill import Workflow

def create_weather_workflow():
    workflow = Workflow(name="weather_query_workflow")

    # 创建节点实例
    input_validation = InputValidationNode()
    weather_query = WeatherQueryNode()
    error_handler = ErrorHandlerNode()

    # 添加节点到工作流
    workflow.add_node(input_validation)
    workflow.add_node(weather_query)
    workflow.add_node(error_handler)

    # 定义边（流程走向）workflow.add_edge(input_validation, weather_query, condition=lambda x: 'error' not in x)
    workflow.add_edge(input_validation, error_handler, condition=lambda x: 'error' in x)

    # 设置入口节点
    workflow.set_entry_point(input_validation)

    return workflow

在实际生产环境中，我们还需要考虑更多因素：