Spring AI Alibaba 03｜Agents 智能体 · 高级特性：控制与流式输出

迭代控制

通过 Hooks 控制 Agent 的执行迭代，防止无限循环或过度成本。

import com.alibaba.cloud.ai.graph.agent.hook.modelcalllimit.ModelCallLimitHook;
import com.alibaba.cloud.ai.graph.checkpoint.savers.MemorySaver;

// 使用内置的 ModelCallLimitHook 限制模型调用次数
ReactAgent agent = ReactAgent.builder()
  .name("my_agent")
  .model(chatModel)
  .hooks(ModelCallLimitHook.builder().runLimit(5).build())  // 限制最多调用 5 次
  .saver(new MemorySaver())
  .build();

import com.alibaba.cloud.ai.graph.agent.hook.ModelHook;
import com.alibaba.cloud.ai.graph.agent.hook.HookPosition;
import com.alibaba.cloud.ai.graph.agent.hook.HookPositions;
import com.alibaba.cloud.ai.graph.agent.hook.JumpTo;
import org.springframework.ai.chat.messages.AssistantMessage;

// 自定义停止条件：基于状态判断是否继续
@HookPositions({HookPosition.BEFORE_MODEL})
public class CustomStopConditionHook extends ModelHook {

  @Override
  public String getName() {
      return "custom_stop_condition";
  }

  @Override
  public CompletableFuture<Map<String, Object>> beforeModel(OverAllState state, RunnableConfig config) {
      // 检查是否找到答案，展示使用 OverAllState
      boolean answerFound = (Boolean) state.value("answer_found").orElse(false);
      // 检查错误次数，展示使用 RunnableConfig
      int errorCount = (Integer) config.context().get("error_count").orElse(0);

      // 找到答案或错误过多时停止
      if (answerFound || errorCount > 3) {
          List<Message> messages = new ArrayList<>();
          messages.add(new AssistantMessage(
              answerFound ? "已找到答案，Agent 执行完成。"
                          : "错误次数过多 (" + errorCount + ")，Agent 执行终止。"
          ));
          // the messages will be appended to the original message list context.
          return CompletableFuture.completedFuture(Map.of("messages", messages));
      }

      return CompletableFuture.completedFuture(Map.of());
  }

}

// 使用自定义停止条件
ReactAgent agent = ReactAgent.builder()
  .name("my_agent")
  .model(chatModel)
  .hooks(new CustomStopConditionHook())
  .saver(new MemorySaver())
  .build();

流式输出

在 Agent 场景中，流式输出的核心是处理 StreamingOutput 类型。无论是模型推理、工具调用还是 Hook 节点，输出都统一为这个类型。

使用 OutputType 区分输出类型：

通过 OutputType 枚举可以区分不同节点的输出，以及判断是流式增量内容还是完成输出：

提示

对于 Hook 等通常非流式的节点，直接使用 AGENT_HOOK_FINISHED 读取内容即可
并非所有节点输出都有意义，尤其是 Hook 节点可能产生无效输出，建议通过 OutputType 进行过滤

import reactor.core.publisher.Flux;
import com.alibaba.cloud.ai.graph.NodeOutput;
import com.alibaba.cloud.ai.graph.streaming.StreamingOutput;
import com.alibaba.cloud.ai.graph.streaming.OutputType;

Flux<NodeOutput> stream = agent.stream("复杂任务");
stream.subscribe(
  output -> {
      // 检查是否为 StreamingOutput 类型
      if (output instanceof StreamingOutput streamingOutput) {
          OutputType type = streamingOutput.getOutputType();
          
          // 处理模型推理的流式输出
          if (type == OutputType.AGENT_MODEL_STREAMING) {
              // 流式增量内容，逐步显示
              System.out.print(streamingOutput.message().getText());
          } else if (type == OutputType.AGENT_MODEL_FINISHED) {
              // 模型推理完成，可获取完整响应
              System.out.println("n模型输出完成");
          }
          
          // 处理工具调用完成（目前不支持 STREAMING）
          if (type == OutputType.AGENT_TOOL_FINISHED) {
              System.out.println("工具调用完成: " + output.node());
          }
          
          // 对于 Hook 节点，通常只关注完成事件（如果Hook没有有效输出可以忽略）
          if (type == OutputType.AGENT_HOOK_FINISHED) {
              System.out.println("Hook 执行完成: " + output.node());
          }
      }
  },
  error -> System.err.println("错误: " + error),
  () -> System.out.println("Agent 执行完成")
);

消息类型识别

StreamingOutput.message() 返回的消息有多种类型，需要结合 OutputType 和消息元数据进行区分处理：

import org.springframework.ai.chat.messages.Message;
import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.ai.chat.messages.ToolResponseMessage;

// 结合 OutputType 和消息类型进行处理
if (output instanceof StreamingOutput streamingOutput) {
  OutputType type = streamingOutput.getOutputType();
  Message message = streamingOutput.message();
  
  // 处理模型流式输出
  if (type == OutputType.AGENT_MODEL_STREAMING) {
      if (message instanceof AssistantMessage assistantMessage) {
          // 检查是否为 Thinking 消息
          Object reasoningContent = assistantMessage.getMetadata().get("reasoningContent");
          if (reasoningContent != null && !reasoningContent.toString().isEmpty()) {
              System.out.print("[Thinking] " + reasoningContent);
          } else {
              // 普通模型响应（增量内容）
              System.out.print(assistantMessage.getText());
          }
      }
  }
  // 处理模型输出完成
  else if (type == OutputType.AGENT_MODEL_FINISHED) {
      if (message instanceof AssistantMessage assistantMessage) {
          if (assistantMessage.hasToolCalls()) {
              // 工具调用请求
              assistantMessage.getToolCalls().forEach(toolCall -> {
                  System.out.println("[Tool Call] " + toolCall.name() + ": " + toolCall.arguments());
              });
          } else {
              // 模型完整响应
              System.out.println("n[Model Finished]");
          }
      }
  }
  // 处理工具执行结果
  else if (type == OutputType.AGENT_TOOL_FINISHED) {
      if (message instanceof ToolResponseMessage toolResponse) {
          toolResponse.getResponses().forEach(response -> {
              System.out.println("[Tool Result] " + response.name() + ": " + response.responseData());
          });
      }
  }
}

提示

先判断 OutputType：通过 OutputType 先确定是模型输出、工具输出还是 Hook 输出，再处理具体消息
Thinking 消息：部分模型（如 DeepSeek-R1）支持输出思考过程，通过 metadata.reasoningContent 获取
工具调用：工具调用请求通常在 AGENT_MODEL_FINISHED 阶段出现，此时 hasToolCalls() 返回 true
流式 vs 完成：STREAMING 阶段内容是增量的，FINISHED 阶段可获取完整消息或执行结果

更多关于 Graph 底层流式输出机制，请参考 Graph 流式输出。

本文整理自 Spring AI Alibaba 官方文档《Agents 智能体》：https://java2ai.com/docs/frameworks/agent-framework/tutorials/agents