File size: 3,034 Bytes
3e363e2 | 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 | """Pydantic API contracts kept aligned with the domain result types."""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
AnswerMode = Literal["entity", "relation", "boolean", "count", "aggregate"]
Operation = Literal[
"find_entities",
"find_neighbors",
"find_paths",
"reason",
"verify",
"predict_links",
"explain",
"count",
"exists",
"all",
"any",
"boolean",
"is_supported",
"assert",
]
class QueryRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
text: str | None = None
start_entities: tuple[str, ...] = ()
start_entity_ids: tuple[str, ...] = ()
allowed_relations: tuple[str, ...] = ()
required_relations: tuple[str, ...] = ()
required_relation_sequence: tuple[str, ...] = ()
forbidden_relations: tuple[str, ...] = ()
required_types: tuple[str, ...] = ()
forbidden_types: tuple[str, ...] = ()
required_intermediate_types: tuple[str, ...] = ()
max_hops: int = Field(default=3, ge=0, le=64)
min_hops: int = Field(default=1, ge=0, le=64)
exact_hops: int | None = Field(default=None, ge=0, le=64)
return_paths: bool = True
verify: bool = True
operation: Operation = "find_entities"
answer_mode: AnswerMode = "entity"
target_entity_ids: tuple[str, ...] = ()
class GraphScope(BaseModel):
model_config = ConfigDict(extra="forbid")
tenant_id: str
graph_version: str | None = None
class ReasoningRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
query: QueryRequest
graph_scope: GraphScope
class CorrectionRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
request: ReasoningRequest
human_action: Literal["accepted", "rejected"]
correct_answer_ids: tuple[str, ...] = ()
correct_path_indices: tuple[int, ...] = ()
reason: str | None = None
class PathStepResponse(BaseModel):
source_id: str
relation_id: str
target_id: str
source_type: str | None = None
target_type: str | None = None
edge_attributes: dict[str, str] = Field(default_factory=dict)
class GraphPathResponse(BaseModel):
steps: tuple[PathStepResponse, ...]
score: float
verified: bool
query_compatibility: float | None = None
class EntityAnswerResponse(BaseModel):
entity_id: str
score: float
class ReasoningResponse(BaseModel):
status: Literal[
"supported", "unsupported", "contradicted", "insufficient_evidence", "error"
]
answers: tuple[EntityAnswerResponse, ...]
supporting_paths: tuple[GraphPathResponse, ...]
conflicting_paths: tuple[GraphPathResponse, ...]
confidence: float = Field(ge=0.0, le=1.0)
calibrated: bool
model_version: str
graph_version: str
reasoning_steps: int = Field(ge=0)
metadata: dict[str, object] = Field(default_factory=dict)
relation_answers: tuple[str, ...] = ()
boolean_answer: bool | None = None
aggregate_value: int | float | None = None
|