From e0ab57282f27ba2d468698bcd888f4c22bb0a5c9 Mon Sep 17 00:00:00 2001 From: chamsechan Date: Sun, 25 Oct 2026 20:48:27 +0800 Subject: [PATCH] refactor(nodes): reuse MapPayloads in common nodes --- src/common_nodes/ocr_detect_node.cpp | 8 +- .../structured_json_parse_node.cpp | 75 +++++----- src/common_nodes/text_rule_match_node.cpp | 129 +++++++++--------- .../nodes/test_structured_json_parse_node.cpp | 12 +- .../unit/nodes/test_text_rule_match_node.cpp | 27 ++++ 5 files changed, 137 insertions(+), 114 deletions(-) diff --git a/src/common_nodes/ocr_detect_node.cpp b/src/common_nodes/ocr_detect_node.cpp index 4f044dfc..bbcfbb4f 100644 --- a/src/common_nodes/ocr_detect_node.cpp +++ b/src/common_nodes/ocr_detect_node.cpp @@ -18,11 +18,9 @@ NodeResult Run(const Inputs& inputs, const Models& models) { if (!result.ok()) return NodeResult::Failure(result.failure()); Outputs outputs; outputs.document = std::move(result).value(); - outputs.text.reserve(outputs.document.size()); - for (const auto& document : outputs.document) { - outputs.text.emplace_back(document.req_id, document.sub_id, - document.data.combined_text); - } + outputs.text = MapPayloads( + outputs.document, + [](const OcrDocumentItem& document) { return document.combined_text; }); return NodeResult::Success(std::move(outputs)); } diff --git a/src/common_nodes/structured_json_parse_node.cpp b/src/common_nodes/structured_json_parse_node.cpp index 88a3e40c..a7ce066a 100644 --- a/src/common_nodes/structured_json_parse_node.cpp +++ b/src/common_nodes/structured_json_parse_node.cpp @@ -133,53 +133,44 @@ bool ParseOrExtractJson(const Params& options, const std::string& input, return false; } -NodeResult Run(const Inputs& inputs, - const Params& options) { - const auto* text_items = inputs.text; - StructuredDocumentBatch output_docs; - output_docs.reserve(text_items->size()); - - for (const auto& item : *text_items) { - std::string parsed_json_str; - nlohmann::json parsed_structured = nlohmann::json::object(); - JsonParseStatus status = JsonParseStatus::kOk; - std::string diag; +NodeResult ParseDocument(const Params& options, + const std::string& text) { + std::string parsed_json_str; + nlohmann::json parsed_structured = nlohmann::json::object(); + JsonParseStatus status = JsonParseStatus::kOk; + std::string diag; - bool ok = ParseOrExtractJson(options, item.data, &parsed_json_str, - &parsed_structured, &status, &diag); - if (ok && !options.ValidateStructuredFields(parsed_structured, &diag)) { - ok = false; - status = JsonParseStatus::kFailed; - } + bool ok = ParseOrExtractJson(options, text, &parsed_json_str, + &parsed_structured, &status, &diag); + if (ok && !options.ValidateStructuredFields(parsed_structured, &diag)) { + ok = false; + status = JsonParseStatus::kFailed; + } - if (!ok) { - if (options.failure_policy_ == "fail") { - return NodeResult::Failure( - "JSON parse failed for sample: " + diag, error::kJsonParse); - } else if (options.failure_policy_ == "emit_diagnostic") { - output_docs.emplace_back( - item.req_id, item.sub_id, - JsonDocumentItem(options.fallback_text_, false, - JsonParseStatus::kFailed, diag, - options.fallback_structured_)); - continue; - } else { // configured_fallback - output_docs.emplace_back( - item.req_id, item.sub_id, - JsonDocumentItem(options.fallback_text_, true, - JsonParseStatus::kFallbackApplied, diag, - options.fallback_structured_)); - continue; - } + if (!ok) { + if (options.failure_policy_ == "fail") { + return NodeResult::Failure( + "JSON parse failed for sample: " + diag, error::kJsonParse); + } else if (options.failure_policy_ == "emit_diagnostic") { + return JsonDocumentItem(options.fallback_text_, false, + JsonParseStatus::kFailed, diag, + options.fallback_structured_); + } else { // configured_fallback + return JsonDocumentItem(options.fallback_text_, true, + JsonParseStatus::kFallbackApplied, diag, + options.fallback_structured_); } - - output_docs.emplace_back( - item.req_id, item.sub_id, - JsonDocumentItem(std::move(parsed_json_str), true, status, diag, - std::move(parsed_structured))); } - return NodeResult::Success(std::move(output_docs)); + return JsonDocumentItem(std::move(parsed_json_str), true, status, diag, + std::move(parsed_structured)); +} + +NodeResult Run(const Inputs& inputs, + const Params& options) { + return MapPayloads(*inputs.text, [&options](const std::string& text) { + return ParseDocument(options, text); + }); } auto Spec() { diff --git a/src/common_nodes/text_rule_match_node.cpp b/src/common_nodes/text_rule_match_node.cpp index 945c9453..e6204891 100644 --- a/src/common_nodes/text_rule_match_node.cpp +++ b/src/common_nodes/text_rule_match_node.cpp @@ -90,83 +90,80 @@ struct Inputs { const TextBatch* text = nullptr; }; -NodeResult Run(const Inputs& inputs, const Params& state) { - const auto* text_items = inputs.text; - RuleMatchBatch output_matches; - output_matches.reserve(text_items->size()); - - for (const auto& item : *text_items) { - const std::string& sentence = item.data; - RuleMatchItem result; - // 首个带类别的命中决定结果名称。 - auto record_hit = [&](const std::string& category, const std::string& word, - float score, const std::string& rule_id) { - result.is_hit = 1; - if (!result.category.empty()) return; - result.category = category; - result.matched_word = word; - result.score = score; - result.rule_id = rule_id; - }; - - // 1. 匹配 categories (词表模式) - for (const auto& [category, words] : state.categories) { - for (const auto& w : words) { - if (w.empty()) continue; - if (sentence.find(w) != std::string::npos) { - record_hit(category, w, 1.0f, {}); - result.matches.push_back( - {RuleMatchSource::kKeyword, category, {}, w, 1.0f}); - } +NodeResult MatchSentence(const Params& state, + const std::string& sentence) { + RuleMatchItem result; + // 首个带类别的命中决定结果名称。 + auto record_hit = [&](const std::string& category, const std::string& word, + float score, const std::string& rule_id) { + result.is_hit = 1; + if (!result.category.empty()) return; + result.category = category; + result.matched_word = word; + result.score = score; + result.rule_id = rule_id; + }; + + // 1. 匹配 categories (词表模式) + for (const auto& [category, words] : state.categories) { + for (const auto& w : words) { + if (w.empty()) continue; + if (sentence.find(w) != std::string::npos) { + record_hit(category, w, 1.0f, {}); + result.matches.push_back( + {RuleMatchSource::kKeyword, category, {}, w, 1.0f}); } } + } - // 2. 匹配 rules (结构化规则模式，支持 regex, exact, contains) - for (const auto& rule : state.rules) { - bool rule_matched = false; - std::unordered_map rule_captures; - - if (rule.strategy == "regex") { - std::string diagnostic; - const TextRegexSearchStatus status = - rule.compiled_regex ? rule.compiled_regex->Search( - sentence, &rule_captures, &diagnostic) - : TextRegexSearchStatus::kNotMatched; - if (status == TextRegexSearchStatus::kError) { - return NodeResult::Failure( - "Regex execution failed for rule '" + rule.id + - "': " + diagnostic, - error::kRegexExecutionFailed); - } - rule_matched = status == TextRegexSearchStatus::kMatched; - } else if (rule.strategy == "exact") { - if (sentence == rule.pattern) { - rule_matched = true; - } - } else { // contains - if (sentence.find(rule.pattern) != std::string::npos) { - rule_matched = true; - } + // 2. 匹配 rules (结构化规则模式，支持 regex, exact, contains) + for (const auto& rule : state.rules) { + bool rule_matched = false; + std::unordered_map rule_captures; + + if (rule.strategy == "regex") { + std::string diagnostic; + const TextRegexSearchStatus status = + rule.compiled_regex ? rule.compiled_regex->Search( + sentence, &rule_captures, &diagnostic) + : TextRegexSearchStatus::kNotMatched; + if (status == TextRegexSearchStatus::kError) { + return NodeResult::Failure( + "Regex execution failed for rule '" + rule.id + "': " + diagnostic, + error::kRegexExecutionFailed); } - - if (rule_matched) { - record_hit(rule.category, rule.pattern, rule.score, rule.id); - for (const auto& [k, v] : rule_captures) result.slots[k] = v; - for (const auto& [k, v] : rule.constants) result.slots[k] = v; - result.matches.push_back({RuleMatchSource::kRule, rule.category, - rule.id, rule.pattern, rule.score}); + rule_matched = status == TextRegexSearchStatus::kMatched; + } else if (rule.strategy == "exact") { + if (sentence == rule.pattern) { + rule_matched = true; + } + } else { // contains + if (sentence.find(rule.pattern) != std::string::npos) { + rule_matched = true; } } - if (!result.is_hit && !state.default_category.empty()) { - record_hit(state.default_category, "", state.default_score, {}); - result.slots["raw_query"] = sentence; + if (rule_matched) { + record_hit(rule.category, rule.pattern, rule.score, rule.id); + for (const auto& [k, v] : rule_captures) result.slots[k] = v; + for (const auto& [k, v] : rule.constants) result.slots[k] = v; + result.matches.push_back({RuleMatchSource::kRule, rule.category, rule.id, + rule.pattern, rule.score}); } + } - output_matches.emplace_back(item.req_id, item.sub_id, std::move(result)); + if (!result.is_hit && !state.default_category.empty()) { + record_hit(state.default_category, "", state.default_score, {}); + result.slots["raw_query"] = sentence; } - return NodeResult::Success(std::move(output_matches)); + return result; +} + +NodeResult Run(const Inputs& inputs, const Params& state) { + return MapPayloads(*inputs.text, [&state](const std::string& sentence) { + return MatchSentence(state, sentence); + }); } auto Spec() { diff --git a/tests/unit/nodes/test_structured_json_parse_node.cpp b/tests/unit/nodes/test_structured_json_parse_node.cpp index e4a73c40..b9fb35a0 100644 --- a/tests/unit/nodes/test_structured_json_parse_node.cpp +++ b/tests/unit/nodes/test_structured_json_parse_node.cpp @@ -1,5 +1,6 @@ #include +#include "contracts/error_codes.h" #include "core/alg_context.h" #include "core/common_contracts.h" #include "tests/support/node_probe.h" @@ -109,8 +110,17 @@ TEST(StructuredJsonParseNodeTest, ASSERT_TRUE(failed.Publish("input", TextBatch{{1, 0, R"({"name":"ok","score":1})"}, {1, 1, R"({"name":false,"score":1})"}})); - EXPECT_NE(fail.node->Process(&failed), 0); + EXPECT_EQ(fail.node->Process(&failed), error::kJsonParse); EXPECT_FALSE(failed.Has("result")); + const std::string message = failed.GetErrorMessage(); + EXPECT_NE(message.find("JSON parse failed for sample: Field 'name' type " + "mismatch, expected string"), + std::string::npos) + << message; + const std::string source = "req_id=1, sub_id=1"; + const auto first = message.find(source); + ASSERT_NE(first, std::string::npos) << message; + EXPECT_EQ(message.find(source, first + 1), std::string::npos) << message; } TEST(StructuredJsonParseNodeTest, diff --git a/tests/unit/nodes/test_text_rule_match_node.cpp b/tests/unit/nodes/test_text_rule_match_node.cpp index c2115c8c..156b4dd8 100644 --- a/tests/unit/nodes/test_text_rule_match_node.cpp +++ b/tests/unit/nodes/test_text_rule_match_node.cpp @@ -1,5 +1,6 @@ #include +#include "contracts/error_codes.h" #include "core/alg_context.h" #include "core/common_contracts.h" #include "tests/support/node_probe.h" @@ -142,5 +143,31 @@ TEST(TextRuleMatchNodeTest, EXPECT_EQ((*output)[0].data.is_hit, 0); EXPECT_EQ((*output)[1].data.category, "new"); } + +TEST(TextRuleMatchNodeTest, + RegexExecutionFailureAbortsBatchAndNamesFailedItemOnce) { + SessionContext session; + NodeProbe node; + ASSERT_TRUE(node.Init("text_rule_match", + {{"rules", + {{{"id", "limited"}, + {"strategy", "regex"}, + {"pattern", "(*LIMIT_MATCH=1)(?:a|b)+c"}}}}}, + session, {{"text", "input", ""}}, + {{"matches", "result", ""}})) + << node.diagnostic; + AlgContext context; + ASSERT_TRUE( + context.Publish("input", TextBatch{{4, 1, "plain"}, {4, 6, "ababdc"}})); + EXPECT_EQ(node.node->Process(&context), error::kRegexExecutionFailed); + EXPECT_FALSE(context.Has("result")); + const std::string message = context.GetErrorMessage(); + EXPECT_NE(message.find("Regex execution failed for rule 'limited'"), + std::string::npos); + const std::string source = "req_id=4, sub_id=6"; + const auto first = message.find(source); + ASSERT_NE(first, std::string::npos) << message; + EXPECT_EQ(message.find(source, first + 1), std::string::npos) << message; +} } // namespace } // namespace llm_edgeflow