AI项目博客我们将每个节点实现为一个简单的函数。记住：节点接受状态，执行工作，并返回更新。
读取和分类节点
import com.alibaba.cloud.ai.graph.action.NodeAction;
import com.alibaba.cloud.ai.graph.OverAllState;
import org.springframework.ai.chat.client.ChatClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.List;
import java.util.ArrayList;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
private static final Logger log = LoggerFactory.getLogger(QuickStartExample.class);
// 读取邮件节点
public static class ReadEmailNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
// 在生产环境中，这将连接到您的邮件服务
String emailContent = state.value("email_content")
.map(v -> (String) v)
.orElse("");
log.info("Processing email: {}", emailContent);
List<String> messages = new ArrayList<>();
messages.add("Processing email: " + emailContent);
return Map.of("messages", messages);
}
}
// 分类意图节点
public static class ClassifyIntentNode implements NodeAction {
private final ChatClient chatClient;
public ClassifyIntentNode(ChatClient.Builder chatClientBuilder) {
this.chatClient = chatClientBuilder.build();
}
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
String emailContent = state.value("email_content")
.map(v -> (String) v)
.orElseThrow(() -> new IllegalStateException("No email content"));
String senderEmail = state.value("sender_email")
.map(v -> (String) v)
.orElse("unknown");
// 按需格式化提示，不存储在状态中
String classificationPrompt = String.format("""
分析这封客户邮件并进行分类：
邮件: %s
发件人: %s
提供分类，包括意图、紧急程度、主题和摘要。
意图应该是以下之一: question, bug, billing, feature, complex
紧急程度应该是以下之一: low, medium, high, critical
以JSON格式返回: {"intent": "...", "urgency": "...", "topic": "...", "summary": "..."}
""", emailContent, senderEmail);
// 获取结构化响应
String response = chatClient.prompt()
.user(classificationPrompt)
.call()
.content();
// 解析为 EmailClassification 对象
EmailClassification classification = parseClassification(response);
// 根据分类确定下一个节点
String nextNode;
if ("billing".equals(classification.getIntent()) ||
"critical".equals(classification.getUrgency())) {
nextNode = "human_review";
} else if (List.of("question", "feature").contains(classification.getIntent())) {
nextNode = "search_documentation";
} else if ("bug".equals(classification.getIntent())) {
nextNode = "bug_tracking";
} else {
nextNode = "draft_response";
}
// 将分类作为单个对象存储在状态中
return Map.of(
"classification", classification,
"next_node", nextNode
);
}
/**
* 简化的JSON解析（实际应用中使用Jackson或Gson）
*/
private EmailClassification parseClassification(String jsonResponse) {
EmailClassification classification = new EmailClassification();
// 简单的正则表达式解析
Pattern intentPattern = Pattern.compile(""intent"s*:s*"([^"]+)"");
Pattern urgencyPattern = Pattern.compile(""urgency"s*:s*"([^"]+)"");
Pattern topicPattern = Pattern.compile(""topic"s*:s*"([^"]+)"");
Pattern summaryPattern = Pattern.compile(""summary"s*:s*"([^"]+)"");
Matcher matcher = intentPattern.matcher(jsonResponse);
if (matcher.find()) {
classification.setIntent(matcher.group(1));
}
matcher = urgencyPattern.matcher(jsonResponse);
if (matcher.find()) {
classification.setUrgency(matcher.group(1));
}
matcher = topicPattern.matcher(jsonResponse);
if (matcher.find()) {
classification.setTopic(matcher.group(1));
}
matcher = summaryPattern.matcher(jsonResponse);
if (matcher.find()) {
classification.setSummary(matcher.group(1));
}
// 如果解析失败，设置默认值
if (classification.getIntent() == null) {
classification.setIntent("question");
}
if (classification.getUrgency() == null) {
classification.setUrgency("medium");
}
if (classification.getTopic() == null) {
classification.setTopic("general");
}
if (classification.getSummary() == null) {
classification.setSummary("需要处理的客户邮件");
}
return classification;
}
}搜索和跟踪节点
import com.alibaba.cloud.ai.graph.action.NodeAction;
import com.alibaba.cloud.ai.graph.OverAllState;
import java.util.Map;
import java.util.List;
// 文档搜索节点
public static class SearchDocumentationNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
// 从分类构建搜索查询
EmailClassification classification = state.value("classification")
.map(v -> (EmailClassification) v)
.orElse(new EmailClassification());
String query = classification.getIntent() + " " + classification.getTopic();
try {
// 实现您的搜索逻辑
// 存储原始搜索结果，而不是格式化的文本
List<String> searchResults = List.of(
"通过设置 > 安全 > 更改密码重置密码",
"密码必须至少12个字符",
"包含大写字母、小写字母、数字和符号"
);
log.info("Searching documentation for: {}", query);
return Map.of(
"search_results", searchResults,
"next_node", "draft_response"
);
} catch (Exception e) {
// 对于可恢复的搜索错误，存储错误并继续
log.warn("Search error: {}", e.getMessage());
List<String> errorResult = List.of("搜索暂时不可用: " + e.getMessage());
return Map.of(
"search_results", errorResult,
"next_node", "draft_response"
);
}
}
}
// Bug跟踪节点
public static class BugTrackingNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
// 在您的bug跟踪系统中创建票据
String ticketId = "BUG-12345"; // 将通过API创建
log.info("Created bug ticket: {}", ticketId);
return Map.of(
"search_results", List.of("已创建Bug票据 " + ticketId),
"current_step", "bug_tracked",
"next_node", "draft_response"
);
}
}响应节点
import com.alibaba.cloud.ai.graph.action.NodeAction;
import com.alibaba.cloud.ai.graph.OverAllState;
import org.springframework.ai.chat.client.ChatClient;
import java.util.Map;
import java.util.List;
import java.util.ArrayList;
import java.util.Optional;
import java.util.stream.Collectors;
// 起草回复节点
public static class DraftResponseNode implements NodeAction {
private final ChatClient chatClient;
public DraftResponseNode(ChatClient.Builder chatClientBuilder) {
this.chatClient = chatClientBuilder.build();
}
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
EmailClassification classification = state.value("classification")
.map(v -> (EmailClassification) v)
.orElse(new EmailClassification());
String emailContent = state.value("email_content")
.map(v -> (String) v)
.orElse("");
// 从原始状态数据按需格式化上下文
List<String> contextSections = new ArrayList<>();
Optional<List<String>> searchResults = state.value("search_results")
.map(v -> (List<String>) v);
if (searchResults.isPresent()) {
// 为提示格式化搜索结果
List<String> docs = searchResults.get();
String formattedDocs = docs.stream()
.map(doc -> "- " + doc)
.collect(Collectors.joining("
"));
contextSections.add("相关文档:
" + formattedDocs);
}
Optional<Map<String, Object>> customerHistory = state.value("customer_history")
.map(v -> (Map<String, Object>) v);
if (customerHistory.isPresent()) {
// 为提示格式化客户数据
Map<String, Object> history = customerHistory.get();
contextSections.add("客户等级: " + history.getOrDefault("tier", "standard"));
}
// 使用格式化的上下文构建提示
String draftPrompt = String.format("""
为这封客户邮件起草回复:
%s
邮件意图: %s
紧急程度: %s
%s
指南:
- 专业且有帮助
- 解决他们的具体问题
- 在相关时使用提供的文档
""",
emailContent,
classification.getIntent(),
classification.getUrgency(),
String.join("
", contextSections)
);
String response = chatClient.prompt()
.user(draftPrompt)
.call()
.content();
// 根据紧急程度和意图确定是否需要人工审核
boolean needsReview =
List.of("high", "critical").contains(classification.getUrgency()) ||
"complex".equals(classification.getIntent());
// 路由到适当的下一个节点
String nextNode = needsReview ? "human_review" : "send_reply";
return Map.of(
"draft_response", response, // 仅存储原始响应
"next_node", nextNode
);
}
}
// 人工审核节点
// 注意：在 interruptBefore 模式下，中断是在编译配置中设置的（见 createEmailAgentGraph 方法）。
// 节点本身不需要做任何特殊处理，只需要正常返回状态即可。
// 当执行到此节点前时，Graph 会自动中断，等待人工输入。
public static class HumanReviewNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
EmailClassification classification = state.value("classification")
.map(v -> (EmailClassification) v)
.orElse(new EmailClassification());
// 准备审核数据
Map<String, Object> reviewData = Map.of(
"email_id", state.value("email_id").map(v -> (String) v).orElse(""),
"original_email", state.value("email_content").map(v -> (String) v).orElse(""),
"draft_response", state.value("draft_response").map(v -> (String) v).orElse(""),
"urgency", classification.getUrgency(),
"intent", classification.getIntent(),
"action", "请审核并批准/编辑此响应"
);
log.info("Waiting for human review: {}", reviewData);
// 返回审核数据和下一个节点
// 注意：在 interruptBefore 模式下，此节点在人工输入后才会执行
return Map.of(
"review_data", reviewData,
"status", "waiting_for_review",
"next_node", "send_reply"
);
}
}
// 发送回复节点
public static class SendReplyNode implements NodeAction {
@Override
public Map<String, Object> apply(OverAllState state) throws Exception {
String draftResponse = state.value("draft_response")
.map(v -> (String) v)
.orElse("");
// 与邮件服务集成
log.info("Sending reply: {}...",
draftResponse.length() > 100
? draftResponse.substring(0, 100)
: draftResponse);
return Map.of("status", "sent");
}
}本文整理自 Spring AI Alibaba 官方文档《Graph 工作流编排指南》：https://java2ai.com/docs/frameworks/graph-core/quick-start
