Spaces:
Runtime error
Runtime error
File size: 3,121 Bytes
aef9206 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | from __future__ import annotations
from dataclasses import dataclass
from app.clients.wolfram import (
WolframAlphaClient,
WolframClientError,
WolframQueryResult,
WolframTextPod,
)
from app.schemas.cross_validation import EngineStatus
@dataclass(frozen=True)
class WolframReferenceResolution:
query: str
status: EngineStatus
reference_result: str | None
reason: str
_PREFERRED_POD_TITLES = {
"result",
"solution",
"solutions",
"root",
"roots",
"domain",
"derivative",
"limit",
"exact result",
"alternate forms",
"decimal approximation",
}
def _clean_wolfram_plaintext(plaintext: str) -> str | None:
cleaned = plaintext.strip()
if not cleaned:
return None
# Wolfram retourne parfois plusieurs lignes. La première ligne utile suffit
# pour une référence courte dans le MVP.
useful_lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
if not useful_lines:
return None
return useful_lines[0]
def _extract_reference_result_from_pods(pods: list[WolframTextPod]) -> str | None:
for pod in pods:
title = pod.title.strip().lower()
plaintext = _clean_wolfram_plaintext(pod.plaintext)
if plaintext and title in _PREFERRED_POD_TITLES:
return plaintext
for pod in pods:
plaintext = _clean_wolfram_plaintext(pod.plaintext)
if plaintext:
return plaintext
return None
def resolve_wolfram_reference_from_result(
wolfram_result: WolframQueryResult,
) -> WolframReferenceResolution:
reference_result = _extract_reference_result_from_pods(wolfram_result.pods)
if wolfram_result.success and reference_result:
return WolframReferenceResolution(
query=wolfram_result.query,
status=EngineStatus.OK,
reference_result=reference_result,
reason="wolfram_reference_result_extracted",
)
if wolfram_result.error:
return WolframReferenceResolution(
query=wolfram_result.query,
status=EngineStatus.FAILED,
reference_result=None,
reason="wolfram_query_error",
)
return WolframReferenceResolution(
query=wolfram_result.query,
status=EngineStatus.FAILED,
reference_result=None,
reason="wolfram_reference_result_missing",
)
async def resolve_wolfram_reference(query: str) -> WolframReferenceResolution:
cleaned_query = query.strip()
if not cleaned_query:
return WolframReferenceResolution(
query="",
status=EngineStatus.FAILED,
reference_result=None,
reason="wolfram_query_empty",
)
try:
wolfram_result = await WolframAlphaClient().query(cleaned_query)
except WolframClientError as exc:
return WolframReferenceResolution(
query=cleaned_query,
status=EngineStatus.FAILED,
reference_result=None,
reason=str(exc),
)
return resolve_wolfram_reference_from_result(wolfram_result)
|