Spring AI Alibaba插件开发指南：自定义Graph节点与工具集成

你是否在开发AI应用时遇到流程复杂难以拆分、第三方工具集成繁琐的问题？本文将带你通过Spring AI Alibaba框架，从零开始实现自定义Graph节点与工具集成，解决多步骤AI任务的编排难题。读完本文，你将掌握节点开发、工具注册、工作流设计的全流程技能，轻松构建灵活可控的AI应用。

核心概念与架构

Spring AI Alibaba Graph是一个工作流与多智能体框架，允许开发者将AI任务抽象为节点（Node）和边（Edge），通过有向图（Graph）形式编排复杂流程。相比传统单轮问答模式，该框架支持分支路由、状态管理和异步执行，特别适合处理需要多步骤协作的AI任务。

核心组件

• StateGraph：工作流定义的核心类，负责注册节点和边关系
• NodeAction：节点执行逻辑接口，所有自定义节点需实现此接口
• EdgeAction：边路由逻辑接口，控制节点间的跳转规则
• OverAllState：全局状态对象，在节点间传递共享数据
• ToolNode：工具集成专用节点，负责调用外部API和服务

框架架构如图所示：

开发环境准备

首先需要在项目中引入Spring AI Alibaba依赖。在pom.xml中添加BOM和Graph核心依赖：

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud.ai</groupId>
            <artifactId>spring-ai-alibaba-bom</artifactId>
            <version>1.0.0.2</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <!-- Graph核心依赖 -->
    <dependency>
        <groupId>com.alibaba.cloud.ai</groupId>
        <artifactId>spring-ai-alibaba-graph-core</artifactId>
    </dependency>
    <!-- DashScope模型适配器 -->
    <dependency>
        <groupId>com.alibaba.cloud.ai</groupId>
        <artifactId>spring-ai-alibaba-starter-dashscope</artifactId>
    </dependency>
</dependencies>

官方文档：spring-ai-alibaba-graph-core/README.md

自定义Graph节点开发

节点开发步骤

1. 实现NodeAction接口
2. 定义输入输出参数
3. 实现业务逻辑
4. 注册为Spring Bean

示例：天气查询节点

创建一个获取实时天气的自定义节点，需要实现NodeAction接口：

import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.action.NodeAction;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;

@Component
public class WeatherQueryNode implements NodeAction {
    
    // 定义状态键名
    public static final String CITY_KEY = "city";
    public static final String WEATHER_KEY = "weather_info";
    
    @Override
    public Map<String, Object> apply(OverAllState state) throws Exception {
        // 从全局状态获取城市参数
        String city = state.getValue(CITY_KEY, String.class);
        if (city == null || city.isEmpty()) {
            throw new IllegalArgumentException("城市参数不能为空");
        }
        
        // 调用天气API（实际项目中替换为真实接口）
        String weatherInfo = getWeatherFromApi(city);
        
        // 将结果写入全局状态
        Map<String, Object> result = new HashMap<>();
        result.put(WEATHER_KEY, weatherInfo);
        
        return result;
    }
    
    private String getWeatherFromApi(String city) {
        // 模拟API调用
        return String.format("城市: %s, 天气: 晴朗, 温度: 25°C", city);
    }
}

节点源码路径：spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/action/NodeAction.java

工具集成与ToolNode使用

ToolNode工作原理

ToolNode是框架提供的工具集成专用节点，通过ToolCallback接口适配外部工具，支持动态参数解析和结果处理。其核心功能包括：

• 工具调用解析
• 参数注入
• 结果封装
• 异常处理

ToolNode源码：spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/node/ToolNode.java

集成第三方工具步骤

1. 实现ToolCallback接口
2. 配置ToolNode
3. 在StateGraph中注册
4. 设置输入输出映射

示例：股票查询工具集成

1. 创建工具回调实现

import org.springframework.ai.tool.ToolCallback;
import org.springframework.ai.tool.ToolParameters;
import org.springframework.stereotype.Component;

@Component
public class StockQueryTool implements ToolCallback {
    
    @Override
    public String getName() {
        // 工具名称，用于LLM识别
        return "stock_query";
    }
    
    @Override
    public String getDescription() {
        return "查询股票实时行情，参数：股票代码(stockCode)";
    }
    
    @Override
    public String call(String parametersJson) {
        // 解析参数
        ToolParameters params = ToolParameters.fromJson(parametersJson);
        String stockCode = params.getString("stockCode");
        
        // 调用股票API（实际项目中替换为真实接口）
        return queryStockPrice(stockCode);
    }
    
    private String queryStockPrice(String stockCode) {
        // 模拟股票查询
        return String.format("股票代码: %s, 最新价格: %.2f元, 涨幅: +1.25%%", 
                            stockCode, 156.8);
    }
}

2. 配置ToolNode

import com.alibaba.cloud.ai.graph.node.ToolNode;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.ai.tool.ToolCallback;
import org.springframework.ai.tool.resolution.ToolCallbackResolver;
import java.util.List;

@Configuration
public class ToolConfig {
    
    @Bean
    public ToolNode stockQueryToolNode(
            StockQueryTool stockQueryTool,
            ToolCallbackResolver resolver) {
        
        // 创建ToolNode并注册工具
        return ToolNode.builder()
                .toolCallbacks(List.of(stockQueryTool))
                .toolCallbackResolver(resolver)
                .llmResponseKey("llm_response")
                .outputKey("stock_result")
                .build();
    }
}

工作流编排与状态管理

StateGraph配置示例

将自定义节点和工具节点编排成完整工作流：

import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.OverAllStateFactory;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.AsyncEdgeAction;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
import static com.alibaba.cloud.ai.graph.node.AsyncNode.node_async;
import static com.alibaba.cloud.ai.graph.action.AsyncEdgeAction.edge_async;

@Configuration
public class WorkflowConfig {
    
    @Bean
    public StateGraph weatherStockWorkflow(
            WeatherQueryNode weatherNode,
            ToolNode stockQueryToolNode) throws Exception {
        
        // 定义全局状态工厂
        OverAllStateFactory stateFactory = () -> {
            OverAllState state = new OverAllState();
            // 注册状态键及更新策略
            state.registerKeyAndStrategy(WeatherQueryNode.CITY_KEY, new ReplaceStrategy());
            state.registerKeyAndStrategy(WeatherQueryNode.WEATHER_KEY, new ReplaceStrategy());
            state.registerKeyAndStrategy("stock_code", new ReplaceStrategy());
            state.registerKeyAndStrategy("stock_result", new ReplaceStrategy());
            return state;
        };
        
        // 创建状态图
        return new StateGraph("weather-stock-workflow", stateFactory)
                // 注册节点
                .addNode("weather_node", node_async(weatherNode))
                .addNode("stock_node", node_async(stockQueryToolNode))
                
                // 定义流程
                .addEdge(StateGraph.START, "weather_node")
                .addEdge("weather_node", "stock_node")
                
                // 条件路由示例（根据天气决定是否查询股票）
                .addConditionalEdges("weather_node", 
                        edge_async(state -> {
                            String weather = state.getValue(WeatherQueryNode.WEATHER_KEY, String.class);
                            return weather.contains("雨") ? "cancel" : "continue";
                        }),
                        Map.of(
                                "continue", "stock_node",
                                "cancel", StateGraph.END
                        ))
                .addEdge("stock_node", StateGraph.END);
    }
}

工作流定义示例：spring-ai-alibaba-graph-core/README.md

状态管理最佳实践

1. 键名设计：使用常量定义状态键，避免硬编码
2. 更新策略：根据业务选择合适的策略
   • ReplaceStrategy：覆盖更新
   • AppendStrategy：追加内容
   • MergeStrategy：合并对象
3. 类型安全：使用state.getValue(key, type)获取强类型值
4. 生命周期：明确状态的创建、更新和销毁时机

调试与监控

日志配置

在application.properties中开启详细日志：

# 开启Graph框架日志
logging.level.com.alibaba.cloud.ai.graph=DEBUG
# 开启Spring AI日志
logging.level.org.springframework.ai=INFO
# 工具调用日志
logging.level.org.springframework.ai.tool=DEBUG

观测性配置

集成Arms观测能力，添加依赖：

<dependency>
    <groupId>com.alibaba.cloud.ai</groupId>
    <artifactId>spring-ai-alibaba-starter-arms-observation</artifactId>
</dependency>

配置观测监听器：

@Bean
public CompiledGraph workflowGraph(StateGraph stateGraph, ObservationRegistry registry) throws GraphStateException {
    return stateGraph.compile(CompileConfig.builder()
            .withLifecycleListener(new GraphObservationLifecycleListener(registry))
            .build());
}

观测配置示例：spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/GraphObservationLifecycleListener.java

完整示例与运行测试

控制器调用示例

import com.alibaba.cloud.ai.graph.CompiledGraph;
import com.alibaba.cloud.ai.graph.OverAllState;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.Map;

@RestController
@RequestMapping("/workflow")
public class WorkflowController {
    
    private final CompiledGraph workflowGraph;
    
    public WorkflowController(@Qualifier("weatherStockWorkflow") CompiledGraph workflowGraph) {
        this.workflowGraph = workflowGraph;
    }
    
    @GetMapping("/run")
    public Map<String, Object> runWorkflow(
            @RequestParam String city,
            @RequestParam String stockCode) {
        
        // 创建初始状态
        OverAllState initialState = new OverAllState();
        initialState.setValue(WeatherQueryNode.CITY_KEY, city);
        initialState.setValue("stock_code", stockCode);
        
        // 执行工作流
        OverAllState finalState = workflowGraph.execute(initialState);
        
        // 返回结果
        return Map.of(
                "weather", finalState.getValue(WeatherQueryNode.WEATHER_KEY),
                "stock", finalState.getValue("stock_result")
        );
    }
}

测试命令

启动应用后，使用curl测试：

# 正常流程测试
curl "http://localhost:8080/workflow/run?city=杭州&stockCode=600036"

# 条件路由测试（模拟雨天场景）
# 注：需要修改天气节点代码返回雨天数据
curl "http://localhost:8080/workflow/run?city=上海&stockCode=601318"

总结与进阶

本文介绍了Spring AI Alibaba框架中自定义Graph节点开发和工具集成的核心流程，包括：

1. 节点开发：通过实现NodeAction接口创建业务节点
2. 工具集成：使用ToolNode适配第三方服务
3. 流程编排：通过StateGraph定义工作流
4. 状态管理：使用OverAllState在节点间共享数据

进阶方向：

• 子图嵌套：使用SubGraph实现复杂流程模块化
• 并行执行：通过ParallelNode实现多节点并行处理
• 持久化：配置状态持久化实现断点续跑
• 监控告警：集成Arms实现全链路观测

官方示例：spring-ai-alibaba-graph-core/src/test/java/com/alibaba/cloud/ai/graph/

通过灵活组合节点和工具，你可以构建从简单到复杂的各类AI应用，充分发挥大模型的能力。建议进一步阅读官方文档和源码，探索更多高级特性。

希望本文对你的开发工作有所帮助！如果有任何问题或建议，欢迎在项目仓库提交issue交流。

点赞+收藏+关注，获取更多Spring AI Alibaba开发技巧！下期预告：《多智能体协作与异步工作流设计》