API Reference

Exceptions module

exception langchain_prolog.exceptions.LangChainPrologException(message: str, *args: Any, **kwargs: Any)

Bases: Exception

Base exception class for langchain-prolog with automatic logging.

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

__init__(message: str, *args: Any, **kwargs: Any) → None

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

exception langchain_prolog.exceptions.PrologRuntimeError(message: str, *args: Any, **kwargs: Any)

Bases: LangChainPrologException

Raised when a Prolog execution error occurs.

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

exception langchain_prolog.exceptions.PrologInitializationError(message: str, *args: Any, **kwargs: Any)

Bases: LangChainPrologException

Raised when Prolog initialization fails.

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

exception langchain_prolog.exceptions.PrologValueError(message: str, *args: Any, **kwargs: Any)

Bases: LangChainPrologException

Raised when a value error occurs.

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

exception langchain_prolog.exceptions.PrologFileNotFoundError(message: str, *args: Any, **kwargs: Any)

Bases: LangChainPrologException

Raised when a file is not found.

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

exception langchain_prolog.exceptions.PrologToolException(message: str, *args: Any, **kwargs: Any)

Bases: LangChainPrologException

Raised when a Prolog tool execution error occurs.

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

__init__(message: str, *args: Any, **kwargs: Any) → None

Initialize exception and log it.

Parameters:

• message (str) – Error message

• *args – Additional positional arguments for Exception

• **kwargs – Additional keyword arguments. Special keys: - log_level: Logging level (default: error) - exc_info: Exception info to include in log

Logger Module

Logging configuration for the langchain-prolog library.

class langchain_prolog.logger.LangChainPrologLogger(name: str = 'langchain_prolog')

Bases: object

Logger for the langchain-prolog package.

Initialize logger with default configuration.

Parameters:

name (str) – Logger name, defaults to ‘langchain_prolog’

__init__(name: str = 'langchain_prolog')

Initialize logger with default configuration.

Parameters:

name (str) – Logger name, defaults to ‘langchain_prolog’

setup_file_logging(log_file: str | Path = 'langchain_prolog.log') → None

Setup file logging.

Parameters:

log_file (str | Path) – Path to log file. Defaults to ‘langchain_prolog.log’.

set_level(level: int) → None

Set logging level.

Parameters:

level (int) – Logging level (e.g., logging.DEBUG, logging.INFO)

trace(msg: str, *args: Any, **kwargs: Any) → None

Log at TRACE level.

property debug: Callable

Debug level logging.

property info: Callable

Info level logging.

property warning: Callable

Warning level logging.

property error: Callable

Error level logging.

property critical: Callable

Critical level logging.

Runnable Module

class langchain_prolog.runnable.PrologConfig(*, rules_file: str | ~pathlib.Path | None = None, default_predicate: str | None = None, query_schema: ~typing.Type[~pydantic.main.BaseModel] | None = None, prolog_flags: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: BaseModel

Configuration for the Prolog interpreter.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

rules_file: str | Path | None

default_predicate: str | None

query_schema: Type[BaseModel] | None

prolog_flags: Dict[str, Any]

classmethod validate_rules_file(v: str | Path | None) → Path | None

classmethod validate_default_predicate(v: str | None) → str | None

classmethod validate_prolog_flags(v: Dict[str, Any]) → Dict[str, Any]

class langchain_prolog.runnable.PrologRunnable(prolog_config: PrologConfig | Dict | None = None, *, verbose: bool = False, callbacks: list[BaseCallbackHandler] | BaseCallbackManager | None = None, tags: List[str] | None = None, metadata: Dict[str, Any] | None = None, name: str | None = None, **kwargs: Any)

Bases: Runnable[str | Dict[Any, Any] | BaseModel | None, Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]

A runnable that executes Prolog queries.

Initialize PrologRunnable with configuration.

Parameters:

• prolog_config (PologConfig | dict, optional) – Configuration for the Prolog interpreter

• verbose (bool) – Whether to enable verbose output. Defaults to False.

• callbacks (Callbacks, optional) – Callback functions.

• tags (list[str], optional) – Optional list of tags.

• metadata (dict, optional) – Optional metadata dictionary.

• name (str, optional) – The name of the Runnable. Used for debugging and tracing.

• **kwargs – Additional arguments passed to the Runnable class constructor

Raises:

• PrologValueError – If the configuration is invalid

• PrologFileNotFoundError – If the rules file doesn’t exist

• PrologRuntimeError – If rules can’t be loaded

__init__(prolog_config: PrologConfig | Dict | None = None, *, verbose: bool = False, callbacks: list[BaseCallbackHandler] | BaseCallbackManager | None = None, tags: List[str] | None = None, metadata: Dict[str, Any] | None = None, name: str | None = None, **kwargs: Any) → None

Initialize PrologRunnable with configuration.

Parameters:

• prolog_config (PologConfig | dict, optional) – Configuration for the Prolog interpreter

• verbose (bool) – Whether to enable verbose output. Defaults to False.

• callbacks (Callbacks, optional) – Callback functions.

• tags (list[str], optional) – Optional list of tags.

• metadata (dict, optional) – Optional metadata dictionary.

• name (str, optional) – The name of the Runnable. Used for debugging and tracing.

• **kwargs – Additional arguments passed to the Runnable class constructor

Raises:

• PrologValueError – If the configuration is invalid

• PrologFileNotFoundError – If the rules file doesn’t exist

• PrologRuntimeError – If rules can’t be loaded

name: str | None

The name of the Runnable. Used for debugging and tracing.

property prolog_config: PrologConfig

Get the Prolog configuration.

load_rules(rules_file: str | Path) → None

Load Prolog rules from a file.

Parameters:

rules_file (str | Path) – Path to the Prolog rules file

Raises:

• PrologFileNotFoundError – If rules_file doesn’t exist

• PrologRuntimeError – If rules can’t be loaded or have gramatical errors

classmethod create_schema(predicate_name: str, arg_names: List[str]) → Type[BaseModel]

Create a Pydantic model for a Prolog predicate.

Parameters:

• predicate_name (str) – Name (functor) of the Prolog predicate

• arg_names (list[str]) – List of argument names in order

Returns:

Pydantic model class for the predicate

Return type:

BaseModel

Raises:

PrologValueError – If predicate_name is empty or arg_names contains invalid names

invoke(input: str | Dict[Any, Any] | BaseModel | None, config: RunnableConfig | None = None, **kwargs: Any) → Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError

Execute a Prolog query and return all solutions at once.

Parameters:

• input (str | dict | BaseModel) – Prolog query

• config (RunnableConfig, optional) – Optional langchain runnable configuration

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Returns:

Prolog query result. True, False or a list of dictionaries

Raises:

• PrologValueError – If input type is invalid or schema is missing for a dictionay input

• PrologRuntimeError – If query execution fails

async ainvoke(input: str | Dict[Any, Any] | BaseModel | None, config: RunnableConfig | None = None, **kwargs: Any) → Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError

Execute a Prolog query asynchronously and return all the solutions at once.

Parameters:

• input (str, dict, BaseModel) – Prolog query

• config (RunnableConfig, optional) – langchain runnable configuration

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Returns:

Query result. True, False or a list of dictionaries

Return type:

PrologResult

Raises:

• PrologValueError – If input type is invalid or schema is missing for a dictionay input

• PrologRuntimeError – If query execution fails

stream(input: str | Dict[Any, Any] | BaseModel | None, config: RunnableConfig | None = None, **kwargs: Any) → Iterator[Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]

Execute a Prolog query and yield solutions one by one.

Parameters:

• input (str, dict, BaseModel) – Prolog query

• config (RunnableConfig, optional) – langchain runnable configuration

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Yields:

PrologResult – Query result. True, False or a list of dictionaries

Raises:

• PrologValueError – If input type is invalid or schema is missing for dictionary input

• PrologRuntimeError – If query execution fails

async astream(input: str | Dict[Any, Any] | BaseModel | None, config: RunnableConfig | None = None, **kwargs: Any) → AsyncIterator[Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]

Execute Prolog query asynchronously and yield results one by one.

Parameters:

• input (str, dict, BaseModel) – Prolog query

• config (RunnableConfig, optional) – langchain runnable configuration

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Yields:

PrologResult – Query result. True, False or a list of dictionaries

Raises:

• PrologValueError – If input type is invalid or schema is missing

• PrologRuntimeError – If query execution fails

batch(inputs: List[str | Dict[Any, Any] | BaseModel | None], config: RunnableConfig | List[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → List[Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]

Execute multiple Prolog queries and return all solutions.

Parameters:

• inputs (list[str | dict | BaseModel]) – List of Prolog queries

• config (RunnableConfig | list[RunnableConfig], optional) – Optional langchain runnable configuration(s)

• return_exceptions (bool) – If True, include exceptions as PrologRuntimeError in the results. Defaults to False.

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Returns:

A list of Prolog queries results (True, False or a list of dictionaries). If return_exceptions is True, exception trowing queries return PrologRuntimeError as the result

Raises:

• PrologValueError – If inputs is not a list

• PrologRuntimeError – If execution fails and return_exceptions is False

Todo

Implement batch execution using Prolog built-in concurrency

async abatch(inputs: List[str | Dict[Any, Any] | BaseModel | None], config: RunnableConfig | List[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → List[Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]

Execute multiple Prolog queries asynchronously and return all solutions.

Parameters:

• inputs (list[str | dict | BaseModel]) – List of Prolog queries

• config (list[RunnableConfig], optional) – Optional langchain runnable configuration(s)

• return_exceptions (bool) – If True, include exceptions as PrologRuntimeError in the results. Defaults to False.

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Returns:

List of Prolog queries results (True, False or list of dictionaries). If return_exceptions is True, exception trowing queries return PrologRuntimeError as a result

Raises:

• PrologValueError – If inputs is not a list

• PrologRuntimeError – If execution fails and return_exceptions is False

batch_as_completed(inputs: Sequence[str | Dict[Any, Any] | BaseModel | None], config: RunnableConfig | Sequence[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → Iterator[Tuple[int, Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]]

Execute multiple Prolog queries and yield results one by one.

Parameters:

• inputs (sequence[str | dict | BaseModel]) – Sequence of Prolog queries

• config (RunnableConfig | sequence[RunnableConfig], optional) – Optional configuration(s) for the runnable

• return_exceptions – If True, include exceptions as PrologRuntimeError in results

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Yields:

Tuples with enumerations of Prolog Query solutions. If return_exceptions is True, exception trowing queries yield PrologRuntimeError as the result

Raises:

• PrologValueError – If inputs is not a list

• PrologRuntimeError – If execution fails and return_exceptions is False

async abatch_as_completed(inputs: Sequence[str | Dict[Any, Any] | BaseModel | None], config: RunnableConfig | Sequence[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) → AsyncIterator[Tuple[int, Literal[True] | Literal[False] | List[Dict[Any, Any]] | PrologRuntimeError]]

Execute multiple Prolog queries asynchronously and yield results one by one.

Parameters:

• inputs (sequence[str | dict | BaseModel]) – Sequence of Prolog queries

• config (RunnableConfig | sequence[RunnableConfig], optional) – Optional configuration(s) for the runnable

• return_exceptions – If True, include exceptions as PrologRuntimeError in results

• **kwargs – Additional arguments for Prolog query execution. Supported keyword arguments by janus-swi: inputs, truth_vals

Yields:

Tuple with enumerations of Prolog Query solutions. If return_exceptions is True, exception trowing queries yield PrologRuntimeError as the result

Raises:

• PrologValueError – If inputs is not a list

• PrologRuntimeError – If execution fails and return_exceptions is False

Tool Module

class langchain_prolog.tool.PrologTool(name: str, description: str, prolog_config: PrologConfig | Dict | None = None, prolog_kwargs: Dict | None = None, *, args_schema: Annotated[type[BaseModel] | dict[str, Any] | None, SkipValidation()] = None, return_direct: bool = False, verbose: bool = False, callbacks: list[BaseCallbackHandler] | BaseCallbackManager | None = None, callback_manager: BaseCallbackManager | None = None, tags: list[str] | None = None, metadata: dict[str, Any] | None = None, handle_tool_error: bool | str | Callable[[ToolException], str] | None = False, handle_validation_error: bool | str | Callable[[ValidationError | ValidationError], str] | None = False, response_format: Literal['content', 'content_and_artifact'] = 'content', func: Callable[[...], str] | None, coroutine: Callable[[...], Awaitable[str]] | None = None, prolog: PrologRunnable = None)

Bases: Tool

Tool for executing Prolog queries.

Initialize the Prolog tool.

Parameters:

• name (str) – The name of the tool.

• description (str) – A description of what the tool does.

• prolog_config (PrologConfig | dict) – Configuration for the Prolog interpreter.

• return_direct (bool) – Whether to return the output directly. Defaults to False.

• verbose (bool) – Whether to enable verbose output. Defaults to False.

• callbacks (Callbacks, optional) – Callback functions to be called during tool execution.

• tags (list[str], optional) – Optional list of tags.

• metadata (dict, optional) – Optional metadata dictionary.

• handle_tool_error (bool | str | Callable, optional) – How to handle tool execution errors.

• handle_validation_error (bool | str | Callable, optional) – How to handle input validation errors.

• response_format ("content" | "content_and_artifact") – Format of the tool’s response.

• **kwargs – Additional arguments passed to BaseTool.

__init__(name: str, description: str, prolog_config: PrologConfig | Dict | None = None, prolog_kwargs: Dict | None = None, **kwargs: Any) → None

Initialize the Prolog tool.

Parameters:

• name (str) – The name of the tool.

• description (str) – A description of what the tool does.

• prolog_config (PrologConfig | dict) – Configuration for the Prolog interpreter.

• return_direct (bool) – Whether to return the output directly. Defaults to False.

• verbose (bool) – Whether to enable verbose output. Defaults to False.

• callbacks (Callbacks, optional) – Callback functions to be called during tool execution.

• tags (list[str], optional) – Optional list of tags.

• metadata (dict, optional) – Optional metadata dictionary.

• handle_tool_error (bool | str | Callable, optional) – How to handle tool execution errors.

• handle_validation_error (bool | str | Callable, optional) – How to handle input validation errors.

• response_format ("content" | "content_and_artifact") – Format of the tool’s response.

• **kwargs – Additional arguments passed to BaseTool.

run(tool_input: str | Dict[Any, Any] | BaseModel | None, **kwargs: Any) → Any

Run the tool.

Parameters:

• tool_input – The input to the tool.

• verbose – Whether to log the tool’s progress. Defaults to None.

• start_color – The color to use when starting the tool. Defaults to ‘green’.

• color – The color to use when ending the tool. Defaults to ‘green’.

• callbacks – Callbacks to be called during tool execution. Defaults to None.

• tags – Optional list of tags associated with the tool. Defaults to None.

• metadata – Optional metadata associated with the tool. Defaults to None.

• run_name – The name of the run. Defaults to None.

• run_id – The id of the run. Defaults to None.

• config – The configuration for the tool. Defaults to None.

• tool_call_id – The id of the tool call. Defaults to None.

• kwargs – Keyword arguments to be passed to tool callbacks

Returns:

The output of the tool.

Raises:

ToolException – If an error occurs during tool execution.

description: str

Used to tell the model how/when/why to use the tool.

You can provide few-shot examples as a part of the description.

func: Callable[..., str] | None

The function to run when the tool is called.

coroutine: Callable[..., Awaitable[str]] | None

The asynchronous version of the function.

name: str

The unique name of the tool that clearly communicates its purpose.

args_schema: Annotated[ArgsSchema | None, SkipValidation()]

Pydantic model class to validate and parse the tool’s input arguments.

Args schema should be either:

• A subclass of pydantic.BaseModel.

or - A subclass of pydantic.v1.BaseModel if accessing v1 namespace in pydantic 2 or - a JSON schema dict

return_direct: bool

Whether to return the tool’s output directly.

Setting this to True means that after the tool is called, the AgentExecutor will stop looping.

verbose: bool

Whether to log the tool’s progress.

callbacks: Callbacks

Callbacks to be called during tool execution.

callback_manager: BaseCallbackManager | None

tags: list[str] | None

Optional list of tags associated with the tool. Defaults to None. These tags will be associated with each call to this tool, and passed as arguments to the handlers defined in callbacks. You can use these to eg identify a specific instance of a tool with its use case.

metadata: dict[str, Any] | None

Optional metadata associated with the tool. Defaults to None. This metadata will be associated with each call to this tool, and passed as arguments to the handlers defined in callbacks. You can use these to eg identify a specific instance of a tool with its use case.

handle_tool_error: bool | str | Callable[[ToolException], str] | None

Handle the content of the ToolException thrown.

handle_validation_error: bool | str | Callable[[ValidationError | ValidationErrorV1], str] | None

Handle the content of the ValidationError thrown.

response_format: Literal['content', 'content_and_artifact']

The tool response format. Defaults to ‘content’.

If “content” then the output of the tool is interpreted as the contents of a ToolMessage. If “content_and_artifact” then the output is expected to be a two-tuple corresponding to the (content, artifact) of a ToolMessage.