控制 Agent

痛点：在多轮 Tool 调用时Agent 可能出现无限循环，导致算力成本失控。

通过 Hooks 可以在每一次模型调用后进行拦截、计数或提前终止，从而防止无限循环。

@Test
public void test19 throws GraphRunnerException {
ChatModel chatModel = CreateChatClient.createDashScopeChatModel;ToolCallback chunkTool = FunctionToolCallback.builder)
.description
.inputType
.build;不过，ModelLoggingHook modelLoggingHook = new ModelLoggingHook;ReactAgent modelHook = ReactAgent.builder
.name
.model
.tools
// 配置一次 agent call 最多只能调用 7 次 LLM
.hooks.runLimit.build,modelLoggingHook)
.build;AssistantMessage message = modelHook.call("""
请严格按照顺序完成：
.依次调用 chunkTool 的 apply 方法。从 SearchToolInput 的 query 属性的值从 0 到 7.
.每次只处理一段，不要跳步.
.最终用中文所有内容.
""");System.out.println);}

流式输出

痛点：传统同步调用必须等模型全部返回后才能看到结果，使用者体验差且耗时长。

使用 ReactAgent.stream 能够在模型产生数据的瞬间立即推送，基于 Reactor 的 Flux 实现非阻塞、零等待的实时展示。

subscribe 方法签名

public final Disposable subscribe(
@Nullable Consumer consumer。@Nullable Consumer errorConsumer,@Nullable Runnable completeConsumer) {
return subscribe null);}

• consumer：正常数据消费回调。
• errorConsumer：异常处理回调。
• completeConsumer：流结束时的回调。

示例 1：全量 & 增量输出演示

该示例演示了所有 OutputType的增量与完成事件。

OutputType	说明
AGENT_MODEL_STREAMING	模型推理的流式增量内容
AGENT_MODEL_FINISHED	模型推理完成。可获取全量内容
AGENT_TOOL_STREAMING	工具调用的流式增量内容
AGENT_TOOL_FINISHED	工具调用完成事件
AGENT_HOOK_STREAMING	Hook 节点的流式增量内容
AGENT_HOOK_FINISHED	Hook 节点执行完成事件

public void test20 throws GraphRunnerException,InterruptedException {
// 创建 tool
ToolCallback searchTool = FunctionToolCallback.builder)
.description
.inputType
.build;ChatModel chatModel = CreateChatClient.createDashScopeChatModel;ReactAgent metaDataTool = ReactAgent.builder
.name
.model
.tools
.hooks)
.build;Flux stream = metaDataTool.stream;CountDownLatch latch = new CountDownLatch;stream.subscribe(
output -> {
if {
OutputType type = streamingOutput.getOutputType;switch {
case AGENT_MODEL_STREAMING ->
System.out.print.getText);case AGENT_MODEL_FINISHED ->
System.out.println;case AGENT_TOOL_STREAMING ->
System.out.println.getText);case AGENT_TOOL_FINISHED ->
System.out.println);case AGENT_HOOK_STREAMING ->
System.out.println.getText);case AGENT_HOOK_FINISHED ->
System.out.println);}
}
},err -> System.err.println。-> latch.countDown
);latch.await，}

示例 2：捕获 Thinking 消息

This example shows how to extract internal “thinking” content that some LLMs emit during reasoning.

public void test21 throws GraphRunnerException,InterruptedException {
ToolCallback searchTool = FunctionToolCallback.builder)
.description
.inputType
.build;ChatModel openAIModel = CreateChatClient.createOpenAIModel;// DeepSeek deepseek-v4-flash
ReactAgent thinkingAgent = ReactAgent.builder
.model
.name
.tools
.build;说起来，Flux stream = thinkingAgent.stream;CountDownLatch latch = new CountDownLatch;stream.subscribe(
output -> {
if {
OutputType type = so.getOutputType;
Message msg = so.message;if {
Object reasoningContent = am.getMetadata.get;if .isEmpty) {
System.out.print;} else {
System.out.print);// 普通增量响应
}
} else if {
if ) {
am.getToolCalls.forEach(tc ->
System.out.println + ": " + tc.arguments));} else {
System.out.println;}
} else if {
tr.getResponses.forEach(r ->
System.out.println + ": " + r.responseData));}
}
},err -> System.err.println。-> latch.countDown
);latch.await，}

🎯

ReactAgent 的主要能力

• Reasoning： 分析问题并决定接下来行动。
• Acting： # 调用 Tools 或直接生成答案。
• Observation： # 接收工具执行结果并写入 State。
• # 基于最新 State 重复 Reason‑Act‑Observe，直至任务完成。

The above loop enables an Agent to decompose complex tasks into manageable sub‑steps and dynamically adapt its strategy using external tools.

Graphs —— 描述非固定执行流程

The execution path of an Agent is decided by LLM at runtime。which means we cannot rely on static pipelines. A "Graph",consisting of Edges,Nodes and a shared State,provides necessary scaffolding:

• E​dge： # 表示 Node 间的转移关系。包括固定边和"条件边"。条件边依据 State 内容动态选择下一条方法，是实现自主决策的关键。
• N​ode： # 单一执行单元，如 Model 调用、Tools、Memory 查询、End、人工干预等。
• S​tate： # 所有节点共享的数据容器，通常是 OverAllState 类。State 决定了 Edge 的走向，也记录每一步产生的信息。

关键组件速览

{% raw %}{% endraw %}{% endraw %}{% raw %}

Name	Description{% raw %}	Model 配置模型参数与生成行为。 Tools 外部功能包装，可在 Agent 中被主动调用。 Interceptor ModelInterceptor 与 ToolInterceptor 用于统一前后处理。 SystemPrompt 程序级指令，用于设定模型角色与行为规范。 RunnableConfig 运行时配置，如 threadId 区分会话、元信息注入等。 Output / OutputSchema 结构化输出定义。前者固定格式，后者更灵活可自定义。 Memory 持久化上下文，实现跨轮记忆。 Hook ModelHook 与 AgentHook 用于日志、提前终止、结果汇总等控制点。 Streaming Output 实时监控整个执行链路，实现低延迟交互。按理说，{% endraw %} 如何利用 Hook 与 Streaming 解决掉实际问题？ 🔥Avoid Infinite Loops： ModelCallLimitHook.runLimit + 实时计数 Hook 可在达到阈值后强制中止循环，从根本上杜绝成本失控。 📦No‑Wait Responses： reactAgent.stream + Flux →​subscribe 能让前端 UI 在首个 token 出现即渲染，明显提高使用者满意度。• Error Handling: 错误回调让异常快速定位，不必等到完整请求结束才发现问题。怎么说呢，• Cumulative Summary: 使用 AGENT_HOOK_FINISHED 收集所有增量后统一生成摘要。实现“一键”， ⚠️Tuning Transparency： LoggingHook / ModelLoggingHook+Streaming 输出可以把每一步 Reason‑Act‑Observe 完整记录下来为调优提供完整审计轨迹。 ⚙️Simplify Debugging： AGENT_TOOL_*。AGENT_MODEL_*,`AGENTHOOK*`` 类型区分，使得在日志或监控网站上能够快速定位是模型卡住还是 Tool 出错。<\/ul> © 2026 Generated by ChatGPT – All rights reserved.

---