構建定時 Agent，基于 Spring AI Alibaba 實現自主運行的人機協同智能 Agent
作者：千習
參考 Langchain 的這篇文章 https://blog.langchain.com/introducing-ambient-agents/，面對目前我們常見 Chat 模式的 Agent 形態以外，還可以充分發揮 Agent 新的運行形態：
- 自主持續運行 Agent：突破目前大部分 AI 應用依賴用戶主動發起對話（Chat）聊天模式的限制。Agent 通過監聽環境信號（如：定時運行、消息事件、上下文變化），Agent 作為智能體，按計劃定時運行的特性在實際業務場景中非常有用。
- Agent 發起人機交互：關鍵點在于部分任務需通過“人工確認”機制（如通知、提問、審核）確保與用戶目標一致，Agent 在自主運行過程中需要一個人機交互窗口，支持人工介入機制，從而在自主性與可控性間取得平衡。
Spring AI Alibaba（以下簡稱 SAA）為開發上述形態的 Agent 提供了相應的框架支持，方便業務快速實現上述場景需求下的 Agent 開發。
為什么需要自主運行 Agent
通過自主運行的 Agent 可以充分發揮其執行效率，讓其具備按預設規劃自主工作的能力，無需每次都等待人類指令運行，可將 Agent 作為一個能替代你部分工作自主運行的“人”來看待。
由此我們可以看到，隨著 Agent 的能力增強，以及 Agent 自主能力的提升，應該會出現以下局面：
- 個體創業者的能力邊界將被大幅度拓寬。
- 在工程領域多人協同的工作可拆解到由一個人+多 Agent 來完成，以提高生產效率。
從業務需求的角度看，可以有以下一些場景可探索挖掘。
- 自動化周期性業務： 面向企業經營的業務系統，可構建周期性地自動收集分析數據、提取數據核心要素、可視化經營報表生成等場景。通過定時運行的 Agent 可實現無人值守的自動化執行，顯著提升數據分析整理效率。相較于傳統經營報表，通過模型對非結構化數據的分析總結能力，可顯著提升報告質量。
- 批量清算處理： 部署多個 Agent，每個 Agent 可同時批量并行處理不同維度的數據信息。在金融行業智能投研領域，AI Agent 可批量處理來自于不同數據源渠道的海量金融數據、新聞資訊和社交媒體信息，借助大模型充分挖掘行業投資機遇和風險事件。
- 事件應急預案響應： 在物聯網、安全監控等場景中，系統需對突發消息事件（如設備報警、用戶行為、危險事件識別觸發）做出自動響應。后臺常駐 Agent 可通過監聽消息隊列或定時巡檢掃描的方式運行，分析識別事件等級做出響應級別的告警消息發布。
- 人類參與決策判斷： Agent 自動處理，人只在需要時進入決策判斷以提升效率。比如 Agent 每天固定時間自動采集匯總供應商的新聞、輿情及社交平臺討論，并進行風險分類與情緒分析；遇到高風險負面事件（如食品安全事故）時，Agent 能通知管理人員人工介入判斷，選擇采取下一步行動策略，比如生成自檢報告、批量通知各網點下架對應供應商產品等。
- 復雜長周期任務： 現階段大批量基于模型進行數據分析處理的場景下，往往執行較慢，就不適合構建 Chat 模式交互。通過任務固定周期觸發、涉及跨域和多模態數據分析，Agent 在后臺將數據做預制加工處理，最終交互上即可提供給用戶加工完成后的數據信息。
- 周期性任務記憶管理： 定時后臺運行的 Agent，在每個周期執行后可以管理其對應的多個周期的執行記憶。通過周期執行記錄信息，可分析隨時間周期變化，借助模型來解讀數據的變化過程。比如出現輿論輿情時，可根據周期變化來分析事件是再持續發酵還是逐步消退，根據變化趨勢來智能路由下一步決策。
如何構建定時運行 Agent
目前市面上定時 AI Task 相關的應用，主要分為以下幾類：
接下來重點說明下 SAA 中的定時 Agent 設計構想，為了支持后臺自主執行的 Agent，在 CompiledGraph 上提供的 schedule(ScheduleConfig config)方法可自由設置該 Agent 定時執行。對于后臺定時運行的 Agent 任務在設計上都通過 ScheduledAgentManager 來進行任務的注冊管理，目前開源默認實現提供了單機應用進程內的 Agent 任務運行管理實現。
通過上述方式，后續可支持在分布式部署場景下的定時 Agent 任務運行，同時為自定義 Agent 任務可視化管理和運行監控提供擴展點。
基于 SAA 自主運行 Agent 展示
Spring AI Alibaba 作為 Agent 開發框架，可方便地為業務開發 Agent 提供了對應解決方案，同時在框架側也提供了構建定時運行的 Agent 能力。后續將通過框架提供的定時調度、人工節點功能，實現兩個實踐案例：店鋪經營分析 Agent、評價輿情分析 Agent。
店鋪經營日報 Agent
通過 SAA 框架可以方便的開發出自定義的 Agent（Workflow Agent），相比于低代碼平臺的流程構建，通過編碼方式定義 Agent 流程會更適合高度定制的場景。比如當前的這個 Agent 我們需要從多個維度（交易訂單、產品說明、客戶畫像信息、門店基礎信息、客戶評價反饋等）去提取對應的數據信息，通過業務編碼再結合 Prompt 模版提供給 LLM 進行分析處理在實現上更具靈活性和可控性。
通過這個方式，讓業務報表既能保持關鍵數據的準確性，又可結合發揮模型對非結構化數據分析，提煉總結出報告的核心要點和下一步行動方案。
參考代碼：
@Bean
public CompiledGraph dailyReportAgent(ChatModel chatModel) throws GraphStateException {
ChatClient chatClient = ChatClient.builder(chatModel).defaultAdvisors(new SimpleLoggerAdvisor()).build();
AsyncNodeAction dataLoaderNode = node_async(
(state) -> {
/*
* 業務報告元數據讀取，如：指定周期的訂單銷量數據、
* 門店產品信息、訂單用戶反饋信息
*/
}
);
// 定義一個經營日報生成節點，根據提供的原始數據信息結合Prompt生成報告
LlmNode llmDataAnalysisNode = LlmNode.builder().chatClient(chatClient)
.paramsKey("data_summary")
.outputKey("summary_message_to_sender")
.userPromptTemplate(DAILY_REPORT)
.build();
// Agent流程：提取原始數據-->生成經營日報-->發送日報信息
StateGraph stateGraph = new StateGraph("OperationAnalysisAgent", () -> {
Map<String, KeyStrategy> strategies = new HashMap<>();
strategies.put("data_summary", new ReplaceStrategy());
strategies.put("summary_message_to_sender", new ReplaceStrategy());
strategies.put("message_sender_result", new ReplaceStrategy());
strategies.put("access_token", new ReplaceStrategy());
return strategies;
}).addNode("data_loader", dataLoaderNode)
.addNode("data_analysis", node_async(llmDataAnalysisNode))
.addNode("message_sender", node_async(generateMessageSender()))
.addEdge(START, "data_loader")
.addEdge("data_loader", "data_analysis")
.addEdge("data_analysis", "message_sender")
.addEdge("message_sender", END);
CompiledGraph compiledGraph = stateGraph.compile();
compiledGraph.setMaxIterations(100);
// 設定當前Agent每天8點執行
ScheduleConfig scheduleConfig = ScheduleConfig.builder()
.cronExpression("0 0 8 */1 * ?") // 每天8點
.build();
compiledGraph.schedule(scheduleConfig);
return compiledGraph;
}
評價輿情分析 Agent
當前案例主要體現 Agent 自主定時運行，僅在分析發現必要時提示用戶進入“人工決策”。相比較于傳統的基于數值比較、關鍵詞匹配等監控方式，通過 LLM 接入可增加業務監控新的維度，讓監控面向更加泛化的數據場景，包括非結構化的文本數據、圖片、影像等等，通過各類型結構數據充分挖掘出潛在風險。
參考代碼：
@Bean
public CompiledGraph evaluationAnalysisAgent(ChatModel chatModel,
FeedbackMapper feedbackMapper) throws GraphStateException {
ChatClient chatClient = ChatClient.builder(chatModel).defaultAdvisors(new SimpleLoggerAdvisor()).build();
EvaluationClassifierNode sessionAnalysis = EvaluationClassifierNode.builder()
.chatClient(chatClient)
.inputTextKey("iterator_item")
.outputKey("session_analysis_result")
.categories(List.of("yes", "no"))
.classificationInstructions(List.of("要求返回純JSON字符串，禁止包含非JSON格式內容，包含字段:user、time、complaint、satisfaction、summary。",
"complaint: 表示當前評價是否為店鋪或產品投訴，取值范圍（yes or no）.",
"satisfaction: 表示用戶實際的消費滿意度",
"summary: 提煉本條核心吐槽點，以及可以改進的方向"))
.build();
StateGraph sessionAnalysisGraph = new StateGraph("session_analysis", subFactory1)
.addNode("iterator", node_async(sessionAnalysis))
.addEdge(StateGraph.START, "iterator")
.addEdge("iterator", StateGraph.END);
AsyncNodeAction sessionLoaderNode = node_async((state) -> {
// 輿情和評價數據加載... ...
return result;
});
// 輿情評價結果分析匯總
AsyncNodeAction sessionResultSummaryNode = node_async((state) -> {
// 輿情評價結果分析匯總... ...
return Map.of();
});
// 通過LLM生成告警報告
LlmNode llmNode = LlmNode.builder().chatClient(chatClient)
.paramsKey("summary_message")
.outputKey("summary_message_to_sender")
.systemPromptTemplate("自定義Prompt")
.build();
// 構建Agent運行流程
StateGraph stateGraph = new StateGraph("ReviewAnalysisAgent", () -> {
Map<String, KeyStrategy> strategies = new HashMap<>();
... ...
return strategies;
}).addNode("session_loader_node", sessionLoaderNode)
.addNode("iteration_session_analysis_node", iterationNode)
.addNode("session_result_summary_node", sessionResultSummaryNode)
.addNode("message_parse", node_async(llmNode))
.addNode("message_sender", node_async(generateMessageSender()))
.addNode("human_feedback", node_async(new HumanFeedbackNode()))
.addNode("human_action", node_async(new HumanActionNode()))
.addEdge(START, "session_loader_node")
.addEdge("session_loader_node", "iteration_session_analysis_node")
.addEdge("iteration_session_analysis_node", "session_result_summary_node")
.addConditionalEdges("session_result_summary_node", AsyncEdgeAction.edge_async(state -> {
Integer complaint = state.value("complaint", 0);
return complaint > 0 ? "message_parse" : StateGraph.END;
}), Map.of("message_parse", "message_parse", StateGraph.END, StateGraph.END))
.addEdge("message_parse", "message_sender")
.addEdge("message_sender", "human_feedback")
.addConditionalEdges("human_feedback", AsyncEdgeAction.edge_async(state -> {
boolean ignore = state.value("ignore", true);
return ignore ? StateGraph.END : "human_action";
}), Map.of("human_action", "human_action", StateGraph.END, StateGraph.END))
.addEdge("message_sender", END);
CompiledGraph compiledGraph = stateGraph.compile();
compiledGraph.setMaxIterations(1000);
// 設定當前Agent每小時執行檢測一次
ScheduleConfig scheduleConfig = ScheduleConfig.builder()
.cronExpression("0 0 */1 * * ?") // 每小時執行檢測一次
.build();
compiledGraph.schedule(scheduleConfig);
return compiledGraph;
}
結束語
自主運行的 AI Agent 開拓了企業智能化場景，通過定時觸發、事件響應和人工協同機制，為業務帶來高效、精準的自動化能力。借助 Spring AI Alibaba 框架，開發者可快速構建定制化 Agent，實現從數據采集、分析到決策的全流程閉環。
相關鏈接：
[1] MSE 任務調度產品
https://mse.console.aliyun.com/#/auth
[2] LangGraph CronJob
https://docs.langchain.com/langgraph-platform/cron-jobs
點擊此處，歡迎針對定時 Agent 運用場景交流溝通。
浙公網安備 33010602011771號