From 168a27c0a1f281ff2c6b636d8f5577b9b9742635 Mon Sep 17 00:00:00 2001 From: "cristiano.hoshikawa" Date: Mon, 7 Sep 2026 22:38:12 -0300 Subject: [PATCH] bugfix reconciliation extractor --- .../routing/enterprise_router.py | 40 ++--- .../agent_framework/runtime/agent_runtime.py | 167 ++++++++++++++---- .../runtime/transaction_parameters.py | 119 ++++++++++++- 3 files changed, 262 insertions(+), 64 deletions(-) diff --git a/agent_framework_oci/libs/agent_framework/src/agent_framework/routing/enterprise_router.py b/agent_framework_oci/libs/agent_framework/src/agent_framework/routing/enterprise_router.py index e622918f..ff6e5d5e 100644 --- a/agent_framework_oci/libs/agent_framework/src/agent_framework/routing/enterprise_router.py +++ b/agent_framework_oci/libs/agent_framework/src/agent_framework/routing/enterprise_router.py @@ -880,12 +880,13 @@ class EnterpriseRouter: text: str, state_decision: RouteDecision, ) -> RouteDecision | None: - """Detecta semanticamente mudança de intenção durante uma transação. + """Detecta abandono explícito da transação seguido (ou não) de nova intenção. - Não existe lista de palavras para desistência ou mudança de assunto. Uma - interrupção nasce de uma intent diferente resolvida por uma keyword - configurada no ``routing.yaml`` ou, na ausência dela, por uma decisão - semântica do LLM com o contexto da transação pendente. + Durante COLLECTING_PARAMETERS/AWAITING_CONFIRMATION, uma classificação + genérica de outra intent nunca basta para interromper a transação. A saída + só é permitida quando o turno expressa semanticamente abandono explícito + do objetivo atual (por exemplo, desistir/deixar para lá) ou handoff tratado + por sua regra própria. Não há blacklist/keyword hardcoded de abandono. """ active_tx = state.get("active_transaction") if isinstance(state.get("active_transaction"), dict) else {} started_intent = str(active_tx.get("started_from_intent") or "").strip() @@ -910,15 +911,10 @@ class EnterpriseRouter: # both failure modes: parameter extraction cannot hide a real new goal, # and a broad keyword cannot steal a legitimate parameter turn. if not (self.enable_llm_router and self.llm is not None): - configured_candidate.metadata = { - **(configured_candidate.metadata or {}), - "transaction_interruption": "intent_shift", - "interrupted_state": state_decision.next_state, - "interrupted_agent": state_decision.agent, - "interrupted_intent": started_intent or previous_intent, - "interruption_source": "configured_routing", - } - return configured_candidate + # Sem classificador semântico não há evidência suficiente para + # distinguir uma nova meta explícita de uma expressão pertencente + # à própria transação (ex.: "isso mesmo, pode cancelar"). + return None if not (self.enable_llm_router and self.llm is not None): return None @@ -951,11 +947,11 @@ class EnterpriseRouter: ), } system = ( - "Você decide apenas se o turno atual continua a transação ativa ou muda de intenção. " - "Use o significado da mensagem e o contexto transacional; não use palavras isoladas como regra. " - "A extração dos parâmetros pendentes já foi tentada antes desta etapa e não consumiu o turno. " - "Se ainda assim a mensagem for apenas uma resposta referencial/valor/nome ao dado pendente, retorne CONTINUE. " - "Se o usuário passou claramente a perseguir outro objetivo, retorne SHIFT e a nova intent permitida. " + "Você decide somente se o usuário ABANDONA EXPLICITAMENTE a transação ativa. " + "Uma nova intent semanticamente plausível NÃO basta. Use o significado completo da mensagem e o contexto transacional; nunca use uma palavra isolada como regra. " + "Retorne CONTINUE quando a fala puder ser confirmação, correção, parâmetro, referência ou instrução ligada à operação ativa, inclusive expressões como 'pode cancelar' quando cancelar pode ser a própria ação em andamento. " + "Retorne ABANDON somente quando houver evidência explícita de desistência/abandono da operação atual; se a mesma fala também trouxer um novo objetivo, informe a nova intent/agent permitidos. " + "Exemplo semântico: 'esquece isso, quero ver minha fatura' = ABANDON + nova intent. " "Retorne somente JSON válido com decision, intent, agent, confidence, reason." ) user = { @@ -981,7 +977,7 @@ class EnterpriseRouter: logger.warning("Falha ao avaliar mudança semântica de intent transacional via LLM: %s", exc) return None - if str(data.get("decision") or "").strip().upper() != "SHIFT": + if str(data.get("decision") or "").strip().upper() != "ABANDON": return None confidence = float(data.get("confidence") or 0.0) if confidence < self.intent_shift_threshold: @@ -1002,11 +998,11 @@ class EnterpriseRouter: reason=str(data.get("reason") or "Mudança semântica de intenção durante transação."), method="llm", metadata={ - "transaction_interruption": "intent_shift", + "transaction_interruption": "explicit_abandonment", "interrupted_state": state_decision.next_state, "interrupted_agent": state_decision.agent, "interrupted_intent": started_intent or previous_intent, - "interruption_source": "semantic_classifier", + "interruption_source": "semantic_abandonment_classifier", "configured_routing_hint": ( configured_candidate.intent if configured_candidate is not None else None ), diff --git a/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/agent_runtime.py b/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/agent_runtime.py index 96deab01..4fa4428f 100644 --- a/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/agent_runtime.py +++ b/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/agent_runtime.py @@ -12,7 +12,7 @@ from typing import Any, Iterable, Mapping from agent_framework.memory.summary_memory import MemoryContext, render_recent_messages -from agent_framework.runtime.transaction_parameters import extract_transaction_parameters, reconcile_transaction_parameters, parse_transaction_confirmation +from agent_framework.runtime.transaction_parameters import extract_current_transaction_parameters, extract_transaction_parameters, reconcile_transaction_parameters, parse_transaction_confirmation from agent_framework.workflows.input_contract import match_expected_input @@ -859,6 +859,17 @@ class AgentRuntimeMixin: if parameter: recovered_arguments = dict(arguments or {}) recovered_arguments.pop(parameter, None) + # Domain validators may declare derived fields that became invalid + # together with the rejected parameter. The framework does not + # interpret field names; it simply honors the generic clear_fields + # contract. This prevents stale derived values from contaminating + # later collection turns. + validator_clear_fields = [ + str(name) for name in (payload.get("clear_fields") or []) + if str(name).strip() + ] + for field in validator_clear_fields: + recovered_arguments.pop(field, None) recovered_policy = self._resolve_tool_execution_policy(tool_name, recovered_arguments) missing = self._missing_required_arguments(recovered_policy, recovered_arguments) if parameter not in missing: @@ -877,6 +888,7 @@ class AgentRuntimeMixin: "status": status, "parameter": parameter, "terminal": False, + "validator_clear_fields": validator_clear_fields, "result": result, } # A recoverable validator may provide a user-facing explanation @@ -1540,36 +1552,56 @@ class AgentRuntimeMixin: arguments: dict[str, Any], reconciliation: dict[str, Any] | None, *, - locked_current_fields: set[str] | None = None, - ) -> tuple[dict[str, Any], list[str], list[str]]: - """Apply temporal reconciliation without mutating explicit current-turn fields. + protected_current_fields: set[str] | None = None, + ) -> tuple[dict[str, Any], list[str]]: + """Apply temporal reconciliation without erasing current-turn candidates. - Values extracted from the current user utterance have highest precedence for - that turn. The temporal reconciler may use them as anchors, but it cannot - clear or overwrite them. A later authoritative pre-validation step may - still reject a current value and trigger the dedicated recovery path. + Current-turn extraction is the first source of truth for *what the user + just said*. Reconciliation may canonicalize that candidate, and business + pre-validation may reject it, but temporal history must not make it + disappear before the validator sees it. """ out = dict(arguments or {}) rec = dict(reconciliation or {}) - locked = {str(name) for name in (locked_current_fields or set())} + protected = {str(name) for name in (protected_current_fields or set())} ignored_clears: list[str] = [] - ignored_values: list[str] = [] - for field in rec.get("clear_fields") or []: name = str(field) - if name in locked: + if name in protected: ignored_clears.append(name) continue out.pop(name, None) + # Canonicalized values are still allowed; this is not a hard field lock. + out.update(dict(rec.get("values") or {})) + return out, ignored_clears - for field, value in dict(rec.get("values") or {}).items(): + @staticmethod + def _apply_prevalidated_structural_candidates( + arguments: dict[str, Any], + reconciliation: dict[str, Any] | None, + policy: dict[str, Any] | None, + ) -> tuple[dict[str, Any], dict[str, str]]: + """Apply structural *candidates* only behind authoritative pre-validation. + + Candidates are not resolved parameters. They may be raw local evidence + fragments and exist solely so a configured domain validator can + canonicalize/reject them. Without enabled pre-validation they are ignored. + """ + out = dict(arguments or {}) + cfg = (policy or {}).get("pre_validation") if isinstance(policy, dict) else None + if not isinstance(cfg, dict) or not cfg.get("enabled") or not str(cfg.get("tool") or "").strip(): + return out, {} + rec = dict(reconciliation or {}) + candidates = dict(rec.get("candidates") or {}) + provenance = dict(rec.get("candidate_provenance") or {}) + applied: dict[str, str] = {} + for field, value in candidates.items(): name = str(field) - if name in locked: - ignored_values.append(name) + if out.get(name) not in _EMPTY_VALUES or value in _EMPTY_VALUES: continue out[name] = value - - return out, ignored_clears, ignored_values + applied[name] = str(provenance.get(name) or "structural_candidate") + return out, applied async def _reconcile_transaction_parameters( self, @@ -1676,7 +1708,7 @@ class AgentRuntimeMixin: route_meta = ((state.get("route_decision") or {}).get("metadata") or {}) if isinstance(state.get("route_decision"), dict) else {} text = route_meta.get("original_input") if route_meta.get("contextual_reentry") else None text = text or state.get("sanitized_input") or state.get("user_text") or "" - return await extract_transaction_parameters( + return await extract_current_transaction_parameters( getattr(self, "llm", None), text=str(text), tool_name=tool_name, @@ -1684,7 +1716,6 @@ class AgentRuntimeMixin: known_arguments=dict(known_arguments or {}), parameter_schema=schema, tool_description=description, - conversational_context="", ) async def _extract_transaction_parameters( @@ -3043,12 +3074,12 @@ class AgentRuntimeMixin: # intent, a transação anterior é encerrada e seus latches são limpos. active_before_interruption = self._active_transaction(state) interruption = str(route_meta.get("transaction_interruption") or "").strip().lower() - if active_before_interruption and interruption == "intent_shift": + if active_before_interruption and interruption == "explicit_abandonment": interrupted_tool = active_before_interruption.get("tool_name") self._finish_active_transaction(state, "CANCELLED") state["transaction_pre_validation"] = None state["tool_policy_result"] = { - "action": "cancelled_by_intent_shift", + "action": "cancelled_by_explicit_abandonment", "tool_name": interrupted_tool, } @@ -3082,21 +3113,59 @@ class AgentRuntimeMixin: policy = self._resolve_tool_execution_policy(tool_name, previous_args) missing_before = self._missing_required_arguments(policy, previous_args) - # Primeira tentativa: coleta tradicional usando SOMENTE a fala - # atual. O histórico não deve participar enquanto a mensagem corrente - # conseguir preencher o contrato normalmente. + # O EnterpriseRouter pode já ter consumido parâmetros pendentes + # neste MESMO turno para garantir precedência da transação sobre + # intent shift. Esses valores não são apenas telemetria: precisam + # ser materializados no estado transacional antes de qualquer nova + # extração. Eles continuam sendo candidatos e passam normalmente + # pela pre-validation autoritativa antes de confirmação/execução. arguments = dict(previous_args) + cached_route_values: dict[str, Any] = {} + cached = route_meta.get("transaction_parameter_values") + missing_before_set = {str(name) for name in missing_before} + if isinstance(cached, dict): + cached_route_values = { + str(key): value + for key, value in cached.items() + if str(key) in missing_before_set and value not in _EMPTY_VALUES + } + arguments.update(cached_route_values) + + # A fala atual pode tanto preencher um campo faltante quanto corrigir + # explicitamente um campo já coletado. Por isso o extrator current-only + # examina TODOS os parâmetros requeridos, nunca apenas missing_parameters. + # Ele continua estritamente local à mensagem atual e só devolve campos + # explicitamente presentes; portanto não compete com a reconciliação + # temporal e não apaga estado quando o usuário não mencionou o campo. + required_fields = [str(name) for name in (policy.get("requires") or [])] + # Legacy/domain overrides historically receive only pending fields. + # Keep that extension contract intact; the base framework extractor + # is the one allowed to inspect every required field so an explicit + # current-turn correction can replace an already collected value. + legacy_override = getattr(type(self), "_extract_transaction_parameters", None) + has_legacy_override = ( + legacy_override is not None + and legacy_override is not AgentRuntimeMixin._extract_transaction_parameters + ) + current_fields = ( + self._missing_required_arguments(policy, arguments) + if has_legacy_override else required_fields + ) extracted_current = await self._extract_transaction_parameters_current_only( state, tool_name=tool_name, - missing_parameters=missing_before, - known_arguments=previous_args, + missing_parameters=current_fields, + known_arguments=arguments, ) arguments.update(dict(extracted_current or {})) state["transaction_parameter_collection"] = { "tool_name": tool_name, "mode": "current_turn", - "resolved_fields": sorted(str(k) for k in (extracted_current or {}).keys()), + "router_resolved_fields": sorted(cached_route_values.keys()), + "current_turn_resolved_fields": sorted(str(k) for k in (extracted_current or {}).keys()), + "resolved_fields": sorted( + set(cached_route_values.keys()) | {str(k) for k in (extracted_current or {}).keys()} + ), } # Segunda opção: somente se a coleta tradicional ainda deixar @@ -3112,11 +3181,13 @@ class AgentRuntimeMixin: parameter_names=required_fields, known_arguments=arguments, ) - locked_current_fields = { - str(k) for k, v in dict(extracted_current or {}).items() if v not in _EMPTY_VALUES + protected_current_fields = { + str(k) + for k, v in dict(extracted_current or {}).items() + if v not in _EMPTY_VALUES } - arguments, ignored_clears, ignored_values = self._apply_transaction_reconciliation( - arguments, reconciliation, locked_current_fields=locked_current_fields + arguments, ignored_clears = self._apply_transaction_reconciliation( + arguments, reconciliation, protected_current_fields=protected_current_fields ) state["transaction_parameter_reconciliation"] = { "tool_name": tool_name, @@ -3125,9 +3196,8 @@ class AgentRuntimeMixin: "provenance": dict(reconciliation.get("provenance") or {}), "clear_fields": list(reconciliation.get("clear_fields") or []), "values": dict(reconciliation.get("values") or {}), - "locked_current_fields": sorted(locked_current_fields), + "protected_current_fields": sorted(protected_current_fields), "ignored_clear_fields": sorted(ignored_clears), - "ignored_reconciled_values": sorted(ignored_values), } else: state.pop("transaction_parameter_reconciliation", None) @@ -3152,6 +3222,12 @@ class AgentRuntimeMixin: tool_name, arguments, state, exclude_fields=policy.get("requires") or [] ) policy = self._resolve_tool_execution_policy(tool_name, arguments) + arguments, applied_candidates = self._apply_prevalidated_structural_candidates( + arguments, locals().get("reconciliation"), policy + ) + if applied_candidates: + state.setdefault("transaction_parameter_reconciliation", {})["applied_structural_candidates"] = applied_candidates + policy = self._resolve_tool_execution_policy(tool_name, arguments) missing = self._missing_required_arguments(policy, arguments) if missing: self._set_collecting_parameters( @@ -3429,11 +3505,13 @@ class AgentRuntimeMixin: parameter_names=required, known_arguments=action_args, ) - locked_current_fields = { - str(k) for k, v in dict(extracted_initial or {}).items() if v not in _EMPTY_VALUES + protected_current_fields = { + str(k) + for k, v in dict(extracted_initial or {}).items() + if str(k) in set(missing_initial) and v not in _EMPTY_VALUES } - action_args, ignored_clears, ignored_values = self._apply_transaction_reconciliation( - action_args, reconciliation, locked_current_fields=locked_current_fields + action_args, ignored_clears = self._apply_transaction_reconciliation( + action_args, reconciliation, protected_current_fields=protected_current_fields ) state["transaction_parameter_reconciliation"] = { "tool_name": selected_action, @@ -3442,9 +3520,8 @@ class AgentRuntimeMixin: "provenance": dict(reconciliation.get("provenance") or {}), "clear_fields": list(reconciliation.get("clear_fields") or []), "values": dict(reconciliation.get("values") or {}), - "locked_current_fields": sorted(locked_current_fields), + "protected_current_fields": sorted(protected_current_fields), "ignored_clear_fields": sorted(ignored_clears), - "ignored_reconciled_values": sorted(ignored_values), } else: state.pop("transaction_parameter_reconciliation", None) @@ -3466,6 +3543,18 @@ class AgentRuntimeMixin: ) state["tool_policy_result"] = {**policy, "tool_name": selected_action} + action_args, applied_candidates = self._apply_prevalidated_structural_candidates( + action_args, locals().get("reconciliation"), policy + ) + if applied_candidates: + state.setdefault("transaction_parameter_reconciliation", {})["applied_structural_candidates"] = applied_candidates + selected = {"tool_name": selected_action, "arguments": action_args} + state["selected_tool_call"] = selected + self._set_active_transaction( + state, tool_name=selected_action, arguments=action_args, status="COLLECTING_PARAMETERS" + ) + policy = self._resolve_tool_execution_policy(selected_action, action_args) + missing = self._missing_required_arguments(policy, action_args) if missing: self._set_collecting_parameters( diff --git a/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/transaction_parameters.py b/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/transaction_parameters.py index 55cc1c37..d6c2687f 100644 --- a/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/transaction_parameters.py +++ b/agent_framework_oci/libs/agent_framework/src/agent_framework/runtime/transaction_parameters.py @@ -69,6 +69,81 @@ def parse_transaction_confirmation(text: str) -> str | None: return None +async def extract_current_transaction_parameters( + llm: Any, + *, + text: str, + tool_name: str, + missing_parameters: list[str], + known_arguments: Mapping[str, Any] | None = None, + parameter_schema: Mapping[str, Any] | None = None, + tool_description: str | None = None, +) -> dict[str, Any]: + """Extract only values explicitly present in the current user utterance. + + This is deliberately narrower than temporal reconciliation: no history, no + cross-turn inference and no routing. Spoken numeric expressions are normalized + according to the declared numeric type, while textual fields must satisfy the + schema description in the current message itself. + """ + names = [str(x) for x in (missing_parameters or []) if str(x).strip()] + message = str(text or "").strip() + if not names or not message or llm is None: + return {} + schema = dict(parameter_schema or {}) + field_spec = { + name: { + "type": schema.get(name, "string") if not isinstance(schema.get(name), dict) else schema.get(name, {}).get("type", "string"), + "description": None if not isinstance(schema.get(name), dict) else schema.get(name, {}).get("description"), + } + for name in names + } + output = {name: None for name in names} + prompt = ( + "Extraia SOMENTE parâmetros explicitamente expressos na mensagem atual do usuário. " + "Não use histórico, não complete referências elípticas com conhecimento anterior e não invente valores. " + "Use exclusivamente type/description do schema para decidir se um trecho preenche um campo. " + "Para campos numéricos, normalize também números falados por extenso para número (por exemplo uma expressão monetária falada deve virar number). " + "Para campos textuais que exigem uma entidade concreta, uma referência apenas por valor/posição não preenche o campo. " + "Retorne SOMENTE JSON válido com exatamente as chaves pedidas; use null quando ausente ou ambíguo.\n" + f"tool: {tool_name}\n" + f"tool_description: {tool_description or ''}\n" + f"known_arguments: {json.dumps(dict(known_arguments or {}), ensure_ascii=False, default=str)}\n" + f"parameter_schema: {json.dumps(field_spec, ensure_ascii=False, default=str)}\n" + f"current_user_message: {message}\n" + f"Formato: {json.dumps(output, ensure_ascii=False)}" + ) + try: + response = await llm.ainvoke( + [{"role": "user", "content": prompt}], + profile_name="transaction_parameter_extraction", + component_name="transaction_parameter_extraction", + generation_name="llm.transaction_parameter_current_only", + temperature=0.0, + ) + except TypeError: + response = await llm.ainvoke([{"role": "user", "content": prompt}]) + except Exception as exc: + logger.warning("transaction.parameter.current_extract_failed tool=%s fields=%s error=%s", tool_name, names, exc) + return {} + raw = _response_text(response).strip() + try: + payload = parse_json_object(raw) + except (TypeError, ValueError): + logger.warning("transaction.parameter.current_extract_invalid tool=%s raw=%r", tool_name, raw[:240]) + return {} + if not isinstance(payload, dict): + return {} + values: dict[str, Any] = {} + for name in names: + value = payload.get(name) + declared = field_spec.get(name, {}).get("type", "string") + coerced = _coerce(value, declared) + if coerced not in _EMPTY_VALUES: + values[name] = coerced + return values + + async def reconcile_transaction_parameters( llm: Any, *, @@ -396,11 +471,49 @@ async def reconcile_transaction_parameters( else: decisions[name] = "unresolved" + # Structural candidates are intentionally NOT resolved values. They are only + # fallback material for a configured authoritative pre-validator. This lets + # the runtime carry a unique local evidence fragment (or, when no anchor + # exists, a single prior user utterance) to domain validation without letting + # the framework invent/canonicalize business entities. + structural_candidates: dict[str, Any] = {} + candidate_provenance: dict[str, str] = {} + # For a field that was already missing before reconciliation, ``clear`` has + # no distinct state effect: there is no prior value to invalidate. Treat it + # as unresolved *only for structural-candidate fallback* so a unique anchored + # history fragment can still be offered to the configured authoritative + # pre-validator. This does not promote the fragment to a resolved value and + # does not change ``clear_fields`` semantics for fields that were previously + # populated. + unresolved_missing = [ + name for name in missing_initial + if decisions.get(name) in {"unresolved", "clear"} and name not in values + ] + anchored = [b for b in evidence_blocks if b.get("anchor_matches") and str(b.get("text") or "").strip()] + prior_user = [ + b for b in evidence_blocks + if int(b.get("priority") or 99) == 4 and str(b.get("text") or "").strip() + ] + for name in unresolved_missing: + declared = str(field_spec.get(name, {}).get("type") or "string").strip().lower() + if declared not in {"string", "str", "text"}: + continue + if len(anchored) == 1: + structural_candidates[name] = str(anchored[0].get("text") or "").strip() + candidate_provenance[name] = "unique_anchored_evidence_block" + logger.info( - "transaction.parameter.reconciled tool=%s decisions=%s resolved=%s clear=%s provenance=%s", - tool_name, decisions, sorted(values), clear_fields, provenance, + "transaction.parameter.reconciled tool=%s decisions=%s resolved=%s clear=%s provenance=%s candidates=%s", + tool_name, decisions, sorted(values), clear_fields, provenance, sorted(structural_candidates), ) - return {"values": values, "decisions": decisions, "provenance": provenance, "clear_fields": clear_fields} + return { + "values": values, + "decisions": decisions, + "provenance": provenance, + "clear_fields": clear_fields, + "candidates": structural_candidates, + "candidate_provenance": candidate_provenance, + } async def extract_transaction_parameters(