# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from dataclasses import dataclass from typing import Any import json import re # Canonical GenLayer Error Prefixes ERROR_EXPECTED = "[EXPECTED]" # Deterministic contract business logic violations ERROR_EXTERNAL = "[EXTERNAL]" # External 4xx errors / client-side failures ERROR_TRANSIENT = "[TRANSIENT]" # Network timeouts / 5xx server issues ERROR_LLM = "[LLM_ERROR]" # Malformed LLM responses or hallucinated schema @allow_storage @dataclass class QueryRecord: """ On-chain persistent representation of a natural language oracle query. """ id: u256 creator: Address question: str status: str # "open" | "resolving" | "resolved" | "failed" reward: u256 # Resolution fee attached to this query (in atto-units) result_json: str # Serialized JSON containing answer_type, answer, confidence, summary, key_sources # ============================================================================= # Standalone Module-Level Helpers (Kept off the Contract class to preserve ABI) # ============================================================================= def _try_parse_number(val: object) -> float | None: """ Attempts to parse a value as a float number, handling numeric strings and commas. Returns None if the value cannot be parsed as a number or is a boolean. """ if val is None or isinstance(val, bool): return None try: s = str(val).strip().replace(",", "") return float(s) except (ValueError, TypeError): return None def _normalize_text(text: str) -> str: """ Normalizes a text string by lowercasing, stripping leading articles, removing punctuation, and collapsing whitespace. """ s = text.strip().lower() # Strip leading common articles s = re.sub(r"^(the|a|an)\s+", "", s) # Remove punctuation s = re.sub(r"[^\w\s]", "", s) # Collapse multiple whitespaces s = re.sub(r"\s+", " ", s).strip() return s def _parse_llm_output(raw_output: str | dict, valid_sources: list[str]) -> dict: """ Defensively parses, sanitizes, and normalizes the LLM JSON response. Guarantees the structured output schema: { "answer_type": "yes_no_unknown" | "numeric" | "short_text", "answer": "yes" | "no" | "unknown" | number | "short string", "confidence": 0-100, "summary": "...", "key_sources": [...] } """ data = raw_output if isinstance(raw_output, str): clean_str = raw_output.strip() # Strip markdown code blocks if present if clean_str.startswith("```"): clean_str = re.sub(r"^```(?:json)?\s*", "", clean_str) clean_str = re.sub(r"\s*```$", "", clean_str) try: data = json.loads(clean_str) except Exception as e: raise gl.vm.UserError(f"{ERROR_LLM} JSON parsing failure: {str(e)}") if not isinstance(data, dict): raise gl.vm.UserError(f"{ERROR_LLM} Output is not a JSON object: {type(data)}") # Key aliasing for raw answer raw_answer = data.get("answer") if raw_answer is None: for alt in ("result", "outcome", "decision", "value"): if alt in data: raw_answer = data[alt] break if raw_answer is None: raise gl.vm.UserError(f"{ERROR_LLM} Missing 'answer' field in response") # Key aliasing for answer_type raw_type = data.get("answer_type") if raw_type is None: for alt in ("type", "response_type", "category"): if alt in data: raw_type = data[alt] break answer_type_str = str(raw_type).strip().lower() if raw_type else "" # Harmonize and validate answer and answer_type num_val = _try_parse_number(raw_answer) if isinstance(raw_answer, bool): final_answer: Any = "yes" if raw_answer else "no" final_type = "yes_no_unknown" elif str(raw_answer).strip().lower() in ("yes", "no", "unknown"): final_answer = str(raw_answer).strip().lower() final_type = "yes_no_unknown" elif answer_type_str == "numeric" or (answer_type_str == "" and num_val is not None): if num_val is not None: final_answer = int(num_val) if num_val.is_integer() else num_val final_type = "numeric" else: final_answer = str(raw_answer).strip() final_type = "short_text" elif answer_type_str == "yes_no_unknown": ans_str = str(raw_answer).strip().lower() if ans_str in ("yes", "no", "unknown"): final_answer = ans_str elif ans_str in ("true", "verified", "correct"): final_answer = "yes" elif ans_str in ("false", "refuted", "incorrect"): final_answer = "no" else: final_answer = "unknown" final_type = "yes_no_unknown" else: # Defaults to short_text final_answer = str(raw_answer).strip() final_type = "short_text" # Confidence coercion raw_conf = data.get("confidence") if raw_conf is None: for alt in ("score", "certainty", "probability"): if alt in data: raw_conf = data[alt] break if raw_conf is None: confidence = 50 else: try: confidence = max(0, min(100, int(round(float(str(raw_conf).strip()))))) except (ValueError, TypeError): confidence = 50 # If answer is unknown, confidence must be forced to <= 30 if final_answer == "unknown": confidence = min(confidence, 30) # Summary coercion raw_summary = data.get("summary") if raw_summary is None: for alt in ("reasoning", "explanation", "analysis", "notes"): if alt in data: raw_summary = data[alt] break summary_str = str(raw_summary).strip() if raw_summary else "Resolution completed based on available evidence." # Key sources: only keep valid URLs that were actually fetched raw_sources = data.get("key_sources") if not isinstance(raw_sources, list): sources_list = [s for s in valid_sources] else: sources_list = [str(s).strip() for s in raw_sources if str(s).strip() and str(s).strip() in valid_sources] if not sources_list and valid_sources: sources_list = [s for s in valid_sources] return { "answer_type": final_type, "answer": final_answer, "confidence": confidence, "summary": summary_str, "key_sources": sources_list, } def _handle_leader_error(leaders_res: object, leader_fn: object) -> bool: """ Canonical error handler for validator disagreement logic. Classifies errors into expected, external, transient, or LLM failure. """ leader_msg = getattr(leaders_res, "message", str(leaders_res)) try: leader_fn() # If validator succeeds while leader errored, disagree to rotate return False except gl.vm.UserError as e: val_msg = getattr(e, "message", str(e)) # Deterministic errors must match exactly if val_msg.startswith(ERROR_EXPECTED) or val_msg.startswith(ERROR_EXTERNAL): return val_msg == leader_msg # Transient errors agree if both are transient if val_msg.startswith(ERROR_TRANSIENT) and leader_msg.startswith(ERROR_TRANSIENT): return True # LLM errors or unknown errors disagree to trigger leader rotation return False except Exception: return False # ============================================================================= # Main Contract # ============================================================================= class NaturalLanguageOracle(gl.Contract): """ NaturalLanguageOracle (NLO) =========================== A reusable GenLayer Intelligent Contract primitive that resolves arbitrary natural-language queries and real-world event questions into structured, consensus-backed on-chain outcomes using mandatory live web evidence and multi-validator GenVM LLM consensus. """ # Storage slots (class-level type declarations) owner: Address min_resolution_fee: u256 next_query_id: u256 queries: TreeMap[u256, QueryRecord] query_evidence: TreeMap[u256, str] query_ids: DynArray[u256] total_resolved: u256 total_failed: u256 resolver_rewards: TreeMap[Address, u256] def __init__(self): """ Initializes the Natural Language Oracle contract. Deployable with 0 constructor arguments for seamless StudioNet compatibility. """ self.owner = gl.message.sender_address self.min_resolution_fee = u256(0) self.next_query_id = u256(1) self.total_resolved = u256(0) self.total_failed = u256(0) # ========================================================================= # Write Methods # ========================================================================= @gl.public.write def create_simple_query(self, question: str, evidence_url: str) -> int: """ Convenience write method to create a query with a single evidence URL. Ideal for direct invocation in the StudioNet interface. :param question: The natural language question to resolve. :param evidence_url: Mandatory evidence URL (http:// or https://). :return: The newly assigned query_id. """ return self.create_query(question, [evidence_url]) @gl.public.write def create_query(self, question: str, evidence_urls: list[str]) -> int: """ Creates a new natural language query for decentralized resolution. Every query about the real world MUST provide at least one valid http/https evidence URL. Anyone can call this and optionally attach a resolution reward (msg.value). :param question: The natural language question to resolve (10-1000 characters). :param evidence_urls: List of 1 to 5 valid http/https evidence URLs. Mandatory. :return: The newly assigned query_id. """ # Validate question cleaned_question = question.strip() if len(cleaned_question) < 10: raise gl.vm.UserError(f"{ERROR_EXPECTED} Question too short: minimum 10 characters required") if len(cleaned_question) > 1000: raise gl.vm.UserError(f"{ERROR_EXPECTED} Question too long: maximum 1000 characters allowed") # Mandatory evidence requirement: queries cannot resolve on internal LLM knowledge alone if not evidence_urls or len(evidence_urls) == 0: raise gl.vm.UserError( f"{ERROR_EXPECTED} Evidence URLs required: at least one valid http/https URL is mandatory" ) if len(evidence_urls) > 5: raise gl.vm.UserError(f"{ERROR_EXPECTED} Too many evidence URLs: maximum 5 permitted") cleaned_urls: list[str] = [] for url in evidence_urls: trimmed_url = url.strip() if not (trimmed_url.startswith("http://") or trimmed_url.startswith("https://")): raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid URL scheme: must start with http:// or https://") if len(trimmed_url) > 512: raise gl.vm.UserError(f"{ERROR_EXPECTED} URL too long: maximum 512 characters allowed") cleaned_urls.append(trimmed_url) if not cleaned_urls: raise gl.vm.UserError( f"{ERROR_EXPECTED} Evidence URLs required: at least one valid http/https URL is mandatory" ) # Enforce minimum fee requirement if set sent_value = gl.message.value if sent_value < self.min_resolution_fee: raise gl.vm.UserError( f"{ERROR_EXPECTED} Insufficient resolution fee: sent {sent_value}, minimum {self.min_resolution_fee} required" ) # Allocate query ID query_id = self.next_query_id self.next_query_id = self.next_query_id + u256(1) self.query_ids.append(query_id) # Store serialized evidence URLs in TreeMap self.query_evidence[query_id] = json.dumps(cleaned_urls) # Persist query state self.queries[query_id] = QueryRecord( id=query_id, creator=gl.message.sender_address, question=cleaned_question, status="open", reward=sent_value, result_json="", ) return int(query_id) @gl.public.write def resolve(self, query_id: int) -> dict: """ Executes multi-validator consensus to resolve a natural language query. Fetches live web evidence, executes LLM evaluation across validators, verifies Equivalence Principle agreement, and commits the result to storage. If no evidence can be successfully fetched (all URLs fail or return empty content), the result is forced to: - answer = "unknown" - confidence <= 30 - summary explaining that no usable evidence was obtained :param query_id: The ID of the query to resolve. :return: The structured consensus result. """ q_id = u256(query_id) if q_id not in self.queries: raise gl.vm.UserError(f"{ERROR_EXPECTED} Query {query_id} does not exist") query = self.queries[q_id] if query.status != "open": raise gl.vm.UserError(f"{ERROR_EXPECTED} Query {query_id} is not open for resolution (status: {query.status})") # Snapshot question and URLs for nondeterministic execution question_text = query.question urls_to_fetch: list[str] = json.loads(self.query_evidence[q_id]) if q_id in self.query_evidence else [] # Define Leader evaluation function def leader_fn() -> dict: """ Fetches live web pages from provided URLs and asks the GenVM LLM to evaluate the question against the factual evidence. Forces answer="unknown" with confidence <= 30 if no evidence is obtained. """ evidence_blocks = [] valid_sources = [] for target_url in urls_to_fetch: try: resp = gl.nondet.web.get(target_url) if resp.status == 200: raw_body = resp.body.decode("utf-8", errors="ignore") # Truncate to 1500 chars to avoid prompt bloat and keep latency low clean_text = raw_body[:1500].replace("\r", " ").replace("\n", " ").strip() if clean_text: evidence_blocks.append(f"Source [{target_url}]:\n{clean_text}") valid_sources.append(target_url) else: evidence_blocks.append(f"Source [{target_url}]: Empty response body") elif 400 <= resp.status < 500: evidence_blocks.append(f"Source [{target_url}]: HTTP {resp.status} Client Error") else: evidence_blocks.append(f"Source [{target_url}]: HTTP {resp.status} Server Error") except Exception: evidence_blocks.append(f"Source [{target_url}]: Network or fetch error") # MANDATORY EVIDENCE RULE: # If no usable evidence could be acquired from any URL, FORCE result to unknown # with confidence <= 30 and NEVER permit an answer based on model internal knowledge. if not valid_sources: return { "answer_type": "yes_no_unknown", "answer": "unknown", "confidence": 20, "summary": "No usable web evidence could be obtained from the provided source URLs.", "key_sources": [], } evidence_prompt_text = "\n\n".join(evidence_blocks) llm_prompt = f"""You are an objective, consensus-driven oracle resolving an externally verifiable question. You MUST evaluate the question based STRICTLY and EXCLUSIVELY on the provided web evidence. DO NOT rely on internal training knowledge. If the provided evidence does not contain sufficient facts to conclusively verify the answer, you MUST set "answer" to "unknown" and "confidence" to 20 or below. QUESTION: "{question_text}" EVIDENCE GATHERED FROM TRUSTED SOURCES: {evidence_prompt_text} INSTRUCTIONS: 1. Determine the appropriate "answer_type": - "yes_no_unknown" for binary yes/no questions, or if evidence is insufficient/inconclusive. - "numeric" if the question asks for a specific number, quantity, count, or measurement. - "short_text" if the question asks for a specific name, entity, date (YYYY-MM-DD), or title. 2. Provide the "answer": - If "answer_type" is "yes_no_unknown": must be "yes", "no", or "unknown". - If "answer_type" is "numeric": provide ONLY the number (e.g. 42 or 3.14). Do NOT include words or units in the answer field. - If "answer_type" is "short_text": provide ONLY the concise canonical name, title, or date (e.g. "argentina", "2024-11-05"). - If the evidence does not clearly prove the answer, answer MUST be "unknown". 3. "confidence": an integer between 0 and 100 reflecting how strongly the provided web evidence proves the answer. Must be <= 30 if answer is "unknown". 4. "summary": 1-2 factual sentences explaining why this answer was reached based on the evidence. 5. "key_sources": JSON array of URLs from the evidence that directly substantiated the conclusion. You must reply with a valid JSON object matching this schema: {{ "answer_type": "yes_no_unknown" | "numeric" | "short_text", "answer": "yes" | "no" | "unknown" | number | "short string", "confidence": 0-100, "summary": "concise factual explanation based on evidence", "key_sources": ["url1", "url2"] }}""" raw_response = gl.nondet.exec_prompt(llm_prompt, response_format="json") return _parse_llm_output(raw_response, valid_sources) # Define Validator evaluation function (Equivalence Principle) def validator_fn(leaders_res: gl.vm.Result) -> bool: """ Independently re-evaluates the question with web evidence and compares substantive findings against the leader's proposal. Enforces strict answer equivalence without substring acceptance. """ if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, leader_fn) try: validator_res = leader_fn() except Exception: # Validator failed to produce valid result, disagree to force rotation return False leader_data = leaders_res.calldata if not isinstance(leader_data, dict) or not isinstance(validator_res, dict): return False # Extract substantive fields l_type = str(leader_data.get("answer_type", "")).strip().lower() v_type = str(validator_res.get("answer_type", "")).strip().lower() l_raw_ans = leader_data.get("answer") v_raw_ans = validator_res.get("answer") # Rule 1: Strict Answer Equivalence (No Substring Matching) # 1a. If either side is "unknown", both MUST be "unknown" l_is_unknown = (str(l_raw_ans).strip().lower() == "unknown") v_is_unknown = (str(v_raw_ans).strip().lower() == "unknown") if l_is_unknown or v_is_unknown: if l_is_unknown != v_is_unknown: return False else: # 1b. Check if both are numeric (either declared as numeric or parseable as numbers) l_num = _try_parse_number(l_raw_ans) v_num = _try_parse_number(v_raw_ans) if l_type == "numeric" or v_type == "numeric" or (l_num is not None and v_num is not None): # One or both are numeric. If either is not parseable as a number, conflict! if l_num is None or v_num is None: return False # Compare numerical values with tight tolerance (prevents "10" vs "100" substring bug) diff = abs(l_num - v_num) if diff > 1e-4: max_val = max(abs(l_num), abs(v_num)) if max_val == 0 or (diff / max_val) > 0.01: return False elif l_type == "yes_no_unknown" or v_type == "yes_no_unknown": # Discrete binary state: exact match required l_ans = str(l_raw_ans).strip().lower() v_ans = str(v_raw_ans).strip().lower() if l_ans != v_ans: return False if l_ans not in ("yes", "no", "unknown"): return False else: # short_text: strict normalized token equality. NO SUBSTRING ACCEPTANCE. l_norm = _normalize_text(str(l_raw_ans)) v_norm = _normalize_text(str(v_raw_ans)) if not l_norm or not v_norm: return False if l_norm != v_norm: # Check token permutation (exact same words, any order) l_tokens = sorted(l_norm.split()) v_tokens = sorted(v_norm.split()) if l_tokens != v_tokens: return False # Rule 2: Confidence Tolerance & Low-Confidence Uncertainty Gate try: l_conf = int(leader_data.get("confidence", 0)) v_conf = int(validator_res.get("confidence", 0)) except (ValueError, TypeError): return False if abs(l_conf - v_conf) > 15: return False # If either side judges confidence <= 30 or "unknown", both must agree on low confidence l_uncertain = (l_conf <= 30) or l_is_unknown v_uncertain = (v_conf <= 30) or v_is_unknown if l_uncertain != v_uncertain: return False # Rule 3: Substantive, non-empty summary (>= 10 chars) l_summary = str(leader_data.get("summary", "")).strip() v_summary = str(validator_res.get("summary", "")).strip() if len(l_summary) < 10 or len(v_summary) < 10: return False # Rule 4: Structural integrity of key_sources if not isinstance(leader_data.get("key_sources"), list): return False return True # Run non-deterministic consensus across validators consensus_result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) # DETERMINISTIC STATE UPDATE (Checks-Effects-Interactions) query.status = "resolved" query.result_json = json.dumps(consensus_result) self.queries[q_id] = query self.total_resolved = self.total_resolved + u256(1) # Distribute resolution reward to caller (resolver) via pull-payment pattern if query.reward > u256(0): resolver = gl.message.sender_address current_reward = self.resolver_rewards[resolver] if resolver in self.resolver_rewards else u256(0) self.resolver_rewards[resolver] = current_reward + query.reward return consensus_result @gl.public.write def claim_reward(self) -> int: """ Allows resolvers to withdraw their accumulated query resolution rewards. Implements secure Checks-Effects pull-payment pattern. """ resolver = gl.message.sender_address amount = self.resolver_rewards[resolver] if resolver in self.resolver_rewards else u256(0) if amount == u256(0): raise gl.vm.UserError(f"{ERROR_EXPECTED} No rewards available for withdrawal") # Effect: reset reward balance before disbursement self.resolver_rewards[resolver] = u256(0) return int(amount) @gl.public.write def update_min_fee(self, new_min_fee: int) -> None: """ Updates the minimum resolution fee required for creating queries. Only the contract owner can call this method. """ if gl.message.sender_address != self.owner: raise gl.vm.UserError(f"{ERROR_EXPECTED} Unauthorized: only contract owner can update fee") self.min_resolution_fee = u256(new_min_fee) # ========================================================================= # View Methods # ========================================================================= @gl.public.view def get_query(self, query_id: int) -> dict: """ Returns full details of a specific query. """ q_id = u256(query_id) if q_id not in self.queries: raise gl.vm.UserError(f"{ERROR_EXPECTED} Query {query_id} does not exist") q = self.queries[q_id] urls = json.loads(self.query_evidence[q_id]) if q_id in self.query_evidence else [] result_parsed = json.loads(q.result_json) if q.result_json else None creator_str = q.creator.as_hex if hasattr(q.creator, "as_hex") else str(q.creator) return { "id": int(q.id), "creator": creator_str, "question": q.question, "evidence_urls": urls, "status": q.status, "reward": int(q.reward), "result": result_parsed, } @gl.public.view def get_result(self, query_id: int) -> dict: """ Returns the resolved result for a query. Reverts if the query is not yet resolved. """ q_id = u256(query_id) if q_id not in self.queries: raise gl.vm.UserError(f"{ERROR_EXPECTED} Query {query_id} does not exist") q = self.queries[q_id] if q.status != "resolved": raise gl.vm.UserError(f"{ERROR_EXPECTED} Query {query_id} is not yet resolved (status: {query.status})") return json.loads(q.result_json) @gl.public.view def get_open_queries(self) -> list[dict]: """ Returns an overview of all currently open queries awaiting resolution. """ open_list: list[dict] = [] for q_id in self.query_ids: q = self.queries[q_id] if q.status == "open": evidence_count = len(json.loads(self.query_evidence[q_id])) if q_id in self.query_evidence else 0 creator_str = q.creator.as_hex if hasattr(q.creator, "as_hex") else str(q.creator) open_list.append({ "id": int(q.id), "creator": creator_str, "question": q.question, "reward": int(q.reward), "evidence_count": evidence_count, }) return open_list @gl.public.view def get_stats(self) -> dict: """ Returns aggregate statistics of the oracle contract. """ total = len(self.query_ids) resolved = int(self.total_resolved) failed = int(self.total_failed) open_count = total - resolved - failed owner_str = self.owner.as_hex if hasattr(self.owner, "as_hex") else str(self.owner) return { "total_queries": total, "open_queries": open_count if open_count >= 0 else 0, "resolved_queries": resolved, "failed_queries": failed, "min_resolution_fee": int(self.min_resolution_fee), "contract_owner": owner_str, } @gl.public.view def get_resolver_reward(self, account: Address) -> int: """ Returns the pending withdrawable reward balance for a resolver address. """ acc = account if isinstance(account, Address) else Address(account) if acc not in self.resolver_rewards: return 0 return int(self.resolver_rewards[acc])