Client and Pipeline API¶
SurveyScribe is the public SVIS extraction facade. It supports synchronous,
asynchronous, and ordered batch conversion. It accepts an injected
StructuredProvider or explicit provider configuration.
The standard output is questionnaire instrument metadata. It is not respondent microdata. Use Completed Questionnaires for caller-defined answer records.
Facade methods¶
| Method | Return | Use |
|---|---|---|
convert(source) |
ExtractionResult[SurveySVIS] |
One source outside an event loop |
aconvert(source) |
ExtractionResult[SurveySVIS] |
One source in async code |
convert_many(sources) |
list[ExtractionResult[SurveySVIS]] |
Ordered synchronous batch |
aconvert_many(sources) |
list[ExtractionResult[SurveySVIS]] |
Ordered async batch |
close() |
None |
Close a synchronous client |
aclose() |
None |
Close an async client |
Use a context manager for lifecycle control. An injected provider cannot be combined with facade-level model, endpoint, credential, API-version, callback, or environment-resolution arguments.
Constructor parameters¶
| Parameter | Purpose |
|---|---|
provider |
Injected StructuredProvider, provider name, or configured default |
model |
Model, deployment, or route identity for facade construction |
api_key |
Explicit primary API key |
base_url |
Explicit HTTPS endpoint; overrides OpenAI-compatible presets |
api_version |
Required Azure-compatible API version |
token_callback |
Synchronous Azure token callback |
config |
Non-secret runtime, generation, retry, routing, and artifact settings |
resolve_environment |
Opt in to SDK environment lookup |
source_registry |
Application source adapter registry |
source_limits |
Resource limits for all facade conversions |
extraction_date |
Stable date override for deterministic output |
from_config() adds an optional explicit TOML path. It checks only
./survey-scribe.toml when no path is supplied.
Errors and result behavior¶
Invalid argument combinations raise ProgrammerInputError. A synchronous call
inside an event loop raises RunningEventLoopError. Use after close raises
ClientClosedError. Configuration errors are raised during construction.
Expected source and provider failures become a failed ExtractionResult with
safe diagnostics. Partial source or provider work can return usable output with
ResultStatus.partial. Cancellation and process-control signals propagate.
SurveyScribe ¶
SurveyScribe(
provider: StructuredProvider | str | None = None,
*,
model: str | None = None,
api_key: str | None = None,
base_url: str | None = None,
api_version: str | None = None,
token_callback: object | None = None,
config: SurveyScribeConfig | None = None,
resolve_environment: bool = False,
source_registry: SourceRegistry | None = None,
source_limits: SourceLimits = DEFAULT_SOURCE_LIMITS,
extraction_date: date | None = None,
)
SVIS-only facade over local source conversion and structured extraction.
from_config
classmethod
¶
from_config(
path: LocalSource | None = None,
*,
config: SurveyScribeConfig | None = None,
resolve_environment: bool = False,
source_registry: SourceRegistry | None = None,
source_limits: SourceLimits = DEFAULT_SOURCE_LIMITS,
) -> Self
Build the facade from explicit config, one TOML path, or opted-in environment.
convert ¶
convert(
source: LocalSource | SourceBundle,
) -> ExtractionResult[SurveySVIS]
Convert one local source from synchronous code.
aconvert
async
¶
aconvert(
source: LocalSource | SourceBundle,
) -> ExtractionResult[SurveySVIS]
Convert one local source through the registry and SVIS pipeline.
convert_many ¶
convert_many(
sources: Iterable[LocalSource | SourceBundle],
) -> list[ExtractionResult[SurveySVIS]]
Convert local sources in stable input order from synchronous code.
aconvert_many
async
¶
aconvert_many(
sources: Iterable[LocalSource | SourceBundle],
) -> list[ExtractionResult[SurveySVIS]]
Convert local sources in input order under one provider-call ceiling.
StructuredPipeline makes one bounded provider call for a caller-defined
Pydantic model. ChunkedStructuredPipeline requires a caller reducer and defines
explicit strict and partial behavior.
Both custom pipelines expose convert(), aconvert(), and extract(). The
response model must be a closed Pydantic BaseModel. The chunked reducer owns
deduplication, conflicts, repeat assembly, and final domain validation.
StructuredPipeline ¶
StructuredPipeline(
provider: StructuredProvider,
response_model: type[T],
*,
source_registry: SourceRegistry | None = None,
source_limits: SourceLimits = DEFAULT_SOURCE_LIMITS,
generation: GenerationConfig | None = None,
retry: RetryConfig | None = None,
max_request_tokens: int = 32000,
max_concurrency: int = 4,
instructions: str | None = None,
)
Bases: Generic[T]
Extract one caller-supplied model with one bounded document-level call.
convert ¶
convert(
source: LocalSource | SourceBundle,
) -> ExtractionResult[T]
Convert one local source outside a running event loop.
aconvert
async
¶
aconvert(
source: LocalSource | SourceBundle,
) -> ExtractionResult[T]
Convert one local source and make exactly one provider call.
extract
async
¶
extract(
document: SourceDocument,
*,
limiter: ConcurrencyLimiter | None = None,
) -> ExtractionResult[T]
Extract an already normalized document without SVIS reconciliation.
ChunkedStructuredPipeline ¶
ChunkedStructuredPipeline(
provider: StructuredProvider,
response_model: type[TChunk],
reducer: Reducer[TChunk, TResult],
*,
source_registry: SourceRegistry | None = None,
source_limits: SourceLimits = DEFAULT_SOURCE_LIMITS,
generation: GenerationConfig | None = None,
retry: RetryConfig | None = None,
max_request_tokens: int = 32000,
overlap_tokens: int = 1000,
max_concurrency: int = 4,
allow_partial: bool = False,
instructions: str | None = None,
)
Bases: Generic[TChunk, TResult]
Extract ordered chunks and combine them only with a caller reducer.
convert ¶
convert(
source: LocalSource | SourceBundle,
) -> ExtractionResult[TResult]
Convert one local source outside a running event loop.
aconvert
async
¶
aconvert(
source: LocalSource | SourceBundle,
) -> ExtractionResult[TResult]
Convert one local source before bounded chunk extraction.
extract
async
¶
extract(
document: SourceDocument,
*,
limiter: ConcurrencyLimiter | None = None,
) -> ExtractionResult[TResult]
Extract chunks, then pass ordered successes and failures to the reducer.
Advanced pipeline exports¶
The module also declares the default extraction contracts and quality helpers as public. Use these only when the facade and custom pipelines do not provide the required control.
pipeline ¶
Deterministic async SVIS extraction and default quality policy.
Reducer
module-attribute
¶
Reducer = Callable[
[
tuple[ProviderResponse[TChunk], ...],
tuple[FailedBlock, ...],
],
TResult | Awaitable[TResult],
]
BlockExtraction ¶
Bases: BaseModel
Strict provider response for one stable source chunk.
model_config
class-attribute
instance-attribute
¶
model_config = ConfigDict(
extra="forbid", frozen=True, hide_input_in_errors=True
)
ExtractedMetadata ¶
Bases: BaseModel
Strict provider response for document-level survey metadata.
model_config
class-attribute
instance-attribute
¶
model_config = ConfigDict(
extra="forbid", frozen=True, hide_input_in_errors=True
)
ExtractedVariable ¶
Bases: BaseModel
One extracted variable bound to exact source blocks in the active chunk.
ExtractionPipeline ¶
ExtractionPipeline(
provider: StructuredProvider,
*,
config: PipelineConfig | None = None,
extraction_date: date | None = None,
)
Extract metadata and variables with one shared global concurrency ceiling.
extract
async
¶
extract(
document: SourceDocument,
*,
limiter: ConcurrencyLimiter | None = None,
) -> ExtractionResult[SurveySVIS]
Extract one normalized document without retaining questionnaire text.
PipelineConfig ¶
Bases: BaseModel
Bounded extraction settings independent from provider SDKs.
model_config
class-attribute
instance-attribute
¶
model_config = ConfigDict(
extra="forbid", frozen=True, hide_input_in_errors=True
)
max_concurrency
class-attribute
instance-attribute
¶
max_concurrency: int = Field(
default=4, ge=1, le=128, strict=True
)
max_request_tokens
class-attribute
instance-attribute
¶
max_request_tokens: int = Field(
default=32000, ge=1024, le=200000, strict=True
)
overlap_tokens
class-attribute
instance-attribute
¶
overlap_tokens: int = Field(
default=1000, ge=0, le=16000, strict=True
)
confidence_threshold
class-attribute
instance-attribute
¶
confidence_threshold: float = Field(
default=0.7, ge=0.0, le=1.0, strict=True
)
generation
class-attribute
instance-attribute
¶
generation: GenerationConfig = Field(
default_factory=GenerationConfig
)
QualityOutcome
dataclass
¶
QualityOutcome(
variables: tuple[SurveyVariable, ...],
diagnostics: tuple[Diagnostic, ...],
failed_records: tuple[FailedBlock, ...] = (),
)
Ordered quality-adjusted variables and stable diagnostics.
QualityRecord
dataclass
¶
QualityRecord(
variable: SurveyVariable,
block_id: str,
source_order: int,
provenance: SourceProvenance,
source_block_ids: tuple[str, ...] = (),
overlap_block_ids: tuple[str, ...] = (),
)
One extracted variable with stable source and overlap provenance.
apply_quality_policy ¶
apply_quality_policy(
records: tuple[QualityRecord, ...],
*,
confidence_threshold: float,
) -> QualityOutcome
Apply the fixed Phase 3 quality table without changing extracted values.
See Custom Structured Output for guarantees, limits, and a credential-free executable example.