langchain_prolog package
Submodules
langchain_prolog.exceptions module
- exception langchain_prolog.exceptions.LangChainPrologException(message: str, *args: Any, **kwargs: Any)[source]
Bases:
ExceptionBase 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[source]
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)[source]
Bases:
LangChainPrologExceptionRaised 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)[source]
Bases:
LangChainPrologExceptionRaised 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)[source]
Bases:
LangChainPrologExceptionRaised 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)[source]
Bases:
LangChainPrologExceptionRaised 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)[source]
Bases:
LangChainPrologExceptionRaised 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[source]
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
langchain_prolog.logger module
Logging configuration for the langchain-prolog library.
- class langchain_prolog.logger.LangChainPrologLogger(name: str = 'langchain_prolog')[source]
Bases:
objectLogger for the langchain-prolog package.
Initialize logger with default configuration.
- Parameters:
name (str) – Logger name, defaults to ‘langchain_prolog’
- __init__(name: str = 'langchain_prolog')[source]
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[source]
Setup file logging.
- Parameters:
log_file (str | Path) – Path to log file. Defaults to ‘langchain_prolog.log’.
- set_level(level: int) None[source]
Set logging level.
- Parameters:
level (int) – Logging level (e.g., logging.DEBUG, logging.INFO)
- 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.
langchain_prolog.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>)[source]
Bases:
BaseModelConfiguration 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.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- rules_file: str | Path | None
- default_predicate: str | None
- query_schema: Type[BaseModel] | None
- prolog_flags: 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)[source]
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
- model_config = {'arbitrary_types_allowed': True}
- __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[source]
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
- property prolog_config: PrologConfig
Get the Prolog configuration.
- load_rules(rules_file: str | Path) None[source]
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][source]
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[source]
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[source]
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][source]
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][source]
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][source]
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][source]
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]][source]
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]][source]
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
langchain_prolog.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, 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', extras: dict[str, Any] | None = None, func: Callable[[...], str] | None, coroutine: Callable[[...], Awaitable[str]] | None = None, prolog: PrologRunnable = None)[source]
Bases:
ToolTool 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[source]
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.
- prolog: PrologRunnable
- run(tool_input: str | Dict[Any, Any] | BaseModel | None, **kwargs: Any) Any[source]
Run the tool.
- Parameters:
tool_input – The input to the tool.
verbose – Whether to log the tool’s progress.
start_color – The color to use when starting the tool.
color – The color to use when ending the tool.
callbacks – Callbacks to be called during tool execution.
tags – Optional list of tags associated with the tool.
metadata – Optional metadata associated with the tool.
run_name – The name of the run.
run_id – The id of the run.
config – The configuration for the tool.
tool_call_id – The id of the tool call.
**kwargs – Keyword arguments to be passed to tool callbacks (event handler)
- Returns:
The output of the tool.
- Raises:
ToolException – If an error occurs during tool execution.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'ignore', 'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
Module contents
langchain_prolog - A LangChain integration for SWI-Prolog
This module provides a bridge between LangChain and SWI-Prolog, allowing seamless integration of Prolog’s logic programming capabilities into LangChain pipelines.
- Key Components:
PrologConfig: Configuration class for Prolog settings
PrologRunnable: Main class for executing Prolog queries
PrologTool: Utility class for integrating Prolog queries into LangChain tools
PrologResult: Type representing possible Prolog query results
PrologInput: Type representing valid input formats
PrologRuntimeError: Exception class for Prolog execution errors
- Requirements:
Python 3.9 or higher
LangChain 0.3.0 or higher
Pydantic 2.0 or higher
SWI-Prolog must be installed and accessible in the system path
On macOS, requires Homebrew installation of SWI-Prolog
The janus_swipl package must be installed
- class langchain_prolog.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>)[source]
Bases:
BaseModelConfiguration 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.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- rules_file: str | Path | None
- default_predicate: str | None
- query_schema: Type[BaseModel] | None
- prolog_flags: Dict[str, Any]
- class langchain_prolog.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)[source]
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
- model_config = {'arbitrary_types_allowed': True}
- __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[source]
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[source]
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][source]
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[source]
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[source]
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][source]
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][source]
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][source]
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][source]
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]][source]
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]][source]
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
- exception langchain_prolog.PrologRuntimeError(message: str, *args: Any, **kwargs: Any)[source]
Bases:
LangChainPrologExceptionRaised 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.PrologFileNotFoundError(message: str, *args: Any, **kwargs: Any)[source]
Bases:
LangChainPrologExceptionRaised 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.PrologInitializationError(message: str, *args: Any, **kwargs: Any)[source]
Bases:
LangChainPrologExceptionRaised 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.PrologToolException(message: str, *args: Any, **kwargs: Any)[source]
Bases:
LangChainPrologExceptionRaised 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[source]
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.PrologValueError(message: str, *args: Any, **kwargs: Any)[source]
Bases:
LangChainPrologExceptionRaised 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
- class langchain_prolog.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, 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', extras: dict[str, Any] | None = None, func: Callable[[...], str] | None, coroutine: Callable[[...], Awaitable[str]] | None = None, prolog: PrologRunnable = None)[source]
Bases:
ToolTool 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[source]
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.
- prolog: PrologRunnable
- run(tool_input: str | Dict[Any, Any] | BaseModel | None, **kwargs: Any) Any[source]
Run the tool.
- Parameters:
tool_input – The input to the tool.
verbose – Whether to log the tool’s progress.
start_color – The color to use when starting the tool.
color – The color to use when ending the tool.
callbacks – Callbacks to be called during tool execution.
tags – Optional list of tags associated with the tool.
metadata – Optional metadata associated with the tool.
run_name – The name of the run.
run_id – The id of the run.
config – The configuration for the tool.
tool_call_id – The id of the tool call.
**kwargs – Keyword arguments to be passed to tool callbacks (event handler)
- Returns:
The output of the tool.
- Raises:
ToolException – If an error occurs during tool execution.
- model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'ignore', 'protected_namespaces': ()}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- 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.
A subclass of pydantic.v1.BaseModel if accessing v1 namespace in pydantic 2
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.
- tags: list[str] | None
Optional list of tags associated with the tool.
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, e.g., identify a specific instance of a tool with its use case.
- metadata: dict[str, Any] | None
Optional metadata associated with the tool.
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, e.g., identify a specific instance of a tool with its usecase.
- 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.
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.
- extras: dict[str, Any] | None
Optional provider-specific extra fields for the tool.
This is used to pass provider-specific configuration that doesn’t fit into standard tool fields.
Example
Anthropic-specific fields like [cache_control](https://docs.langchain.com/oss/python/integrations/chat/anthropic#prompt-caching), [defer_loading](https://docs.langchain.com/oss/python/integrations/chat/anthropic#tool-search), or input_examples.
```python @tool(extras={“defer_loading”: True, “cache_control”: {“type”: “ephemeral”}}) def my_tool(x: str) -> str:
return x
- exception langchain_prolog.ValidationError(title, line_errors, input_type='python', hide_input=False)
Bases:
ValueError- error_count()
- errors(*, include_url=True, include_context=True, include_input=True)
- from_exception_data(line_errors, input_type='python', hide_input=False)
- json(*, indent=None, include_url=True, include_context=True, include_input=True)
- title