Utils

Utils module for Graph Data Store.

GraphitiEMInvokerAdapter(em_invoker)

Bases: EmbedderClient

Adapts GLLM BaseEMInvoker to Graphiti EmbeddingClient interface.

This adapter wraps a BaseEMInvoker instance to provide compatibility with Graphiti's EmbeddingClient interface. It directly returns embeddings from the underlying invoker without any conversion.

Attributes:

Name	Type	Description
_em_invoker	BaseEMInvoker	The underlying EM invoker instance.

Note

The adapter handles batch embedding operations by delegating to the underlying EM invoker.

Initialize the GraphitiEMInvokerAdapter.

Parameters:

Name	Type	Description	Default
em_invoker	BaseEMInvoker	The EM invoker to wrap.	required

create(input_data) async

Create embedding for a single input.

Parameters:

Name	Type	Description	Default
input_data	str | list[str]	Input text string or list of strings.	required

Returns:

Type	Description
list[float]	list[float]: Embedding vector.

Raises:

Type	Description
ValueError	If input_data is not a string or list of strings.
RuntimeError	If the EM invoker returns an empty embeddings list.

create_batch(input_data_list) async

Create embeddings for multiple inputs.

Parameters:

Name	Type	Description	Default
input_data_list	list[str | list[str]]	List of input texts or list of strings.	required

Returns:

Type	Description
list[list[float]]	list[list[float]]: List of embedding vectors.

Raises:

Type	Description
ValueError	If any input is not a string or list of strings.

create_embeddings(texts) async

Create embeddings for a list of texts.

Directly invokes the underlying EM invoker with the list of texts and returns the embeddings without any conversion. The EM invoker is expected to handle batching internally.

Parameters:

Name	Type	Description	Default
texts	list[str]	The list of texts to embed.	required

Returns:

Type	Description
list[list[float]]	list[list[float]]: The list of embedding vectors (each a list of floats) as returned by the underlying EM invoker.

set_tracer(tracer)

Set the tracer for OpenTelemetry tracing.

This method is required by Graphiti's EmbedderClient interface but is a no-op for this adapter as tracing is not currently implemented.

Parameters:

Name	Type	Description	Default
tracer	Any	The OpenTelemetry tracer instance.	required

GraphitiLMInvokerAdapter(lm_invoker)

Bases: LLMClient

Adapts GLLM BaseLMInvoker to Graphiti LLMClient interface.

This adapter wraps a BaseLMInvoker instance to provide compatibility with Graphiti's LLMClient interface. It handles conversion between GLLM's message format and Graphiti's prompt-based generation interface.

Attributes:

Name	Type	Description
_base_lm_invoker	BaseLMInvoker	The underlying LM invoker instance.
_invoker_pool	dict	Cache of invokers per response model.

Initialize the adapter with a base LM invoker.

Parameters:

Name	Type	Description	Default
lm_invoker	BaseLMInvoker	The LM invoker to wrap and configure with response schemas.	required

generate(prompt, response_model) async

Generate a response using the underlying LM invoker.

Converts the prompt to GLLM Message format, invokes the LM invoker, and parses the response into the requested Pydantic model.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt string to send to the LLM.	required
response_model	type[T]	The Pydantic model class to parse the response into.	required

Returns:

Name	Type	Description
T	BaseModel	An instance of response_model parsed from the LLM output.

set_tracer(tracer)

Set the tracer for OpenTelemetry tracing.

This method is required by Graphiti's LLMClient interface but is a no-op for this adapter as tracing is not currently implemented.

Parameters:

Name	Type	Description	Default
tracer	Any	The OpenTelemetry tracer instance.	required

GraphitiRerankerType

Bases: str, Enum

Enum for Graphiti reranker types.

LightRAGEMInvokerAdapter(em_invoker=None, embedding_dim=0, func=None, max_token_size=None, send_dimensions=False, model_name=None, _em_invoker=None, **kwargs) dataclass

Bases: EmbeddingFunc

Adapter for embedding model invokers to work with LightRAG.

This adapter wraps BaseEMInvoker instances to make them compatible with LightRAG's expected interface.

Attributes:

Name	Type	Description
_em_invoker	BaseEMInvoker	The EM invoker to use.
func	callable	The embedding function.
embedding_dim	int	The embedding dimension. Defaults to 0.

Initialize the LightRAGEMInvokerAdapter.

Parameters:

Name	Type	Description	Default
em_invoker	BaseEMInvoker | None	The EM invoker to use.	None
embedding_dim	int	The embedding dimension. Defaults to 0.	0
func	Callable | None	The embedding function. Defaults to None (uses call).	None
max_token_size	int | None	The max token size. Defaults to None.	None
send_dimensions	bool	Whether to send dimensions. Defaults to False.	False
model_name	str | None	The model name. Defaults to None (extracted from invoker).	None
_em_invoker	BaseEMInvoker | None	Internal dataclass field for preserving the invoker through dataclasses.replace(). Use em_invoker instead.	None
**kwargs	Any	Additional keyword arguments.	{}

__call__(input, **kwargs) async

Make the adapter callable for compatibility with LightRAG.

Parameters:

Name	Type	Description	Default
input	str | list[str]	The input text or list of texts to embed.	required
**kwargs	Any	Additional keyword arguments (ignored, for compatibility).	{}

Returns:

Type	Description
ndarray	np.ndarray: The embeddings for the input texts.

__deepcopy__(memo)

Custom deepcopy implementation to handle non-serializable objects.

This method is called when copy.deepcopy() is invoked on this object. We create a new instance without deep-copying the invoker object which may contain non-serializable components.

Parameters:

Name	Type	Description	Default
memo	dict	Memoization dictionary for deepcopy process	required

Returns:

Name	Type	Description
LightRAGEMInvokerAdapter	LightRAGEMInvokerAdapter	A new instance with the same invoker reference

ensure_initialized() async

Ensure that the adapter is initialized.

This asynchronous method ensures that the embedding dimension is determined. If the embedding dimension is 0, it will determine the dimension by calling the embedding invoker with a test input. Raises an error if initialization fails.

Raises:

Type	Description
RuntimeError	If embedding dimension cannot be determined after initialization.

LightRAGLMInvokerAdapter(lm_invoker)

LMInvoker adapter for the LightRAG module.

This adapter is used to adapt the LMInvoker interface to the LightRAG module. It handles the conversion between different prompt formats and manages asynchronous invocation in a way that's compatible with nested event loops.

Initialize the LightRAGLMInvokerAdapter.

Parameters:

Name	Type	Description	Default
lm_invoker	BaseLMInvoker	The LM invoker to use.	required

__call__(prompt, system_prompt=None, history_messages=None, **kwargs) async

Make the adapter callable for compatibility with LightRAG.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt to invoke the LM invoker with.	required
system_prompt	str | None	The system prompt to format in string format. Defaults to None.	None
history_messages	list[dict[str, Any]] | None	The history messages to format in OpenAI format. Defaults to None.	None
**kwargs	Any	Additional keyword arguments for the LM invoker.	{}

Returns:

Name	Type	Description
str	str	The response from the LM invoker.

__deepcopy__(memo)

Custom deepcopy implementation to handle non-serializable objects.

This method is called when copy.deepcopy() is invoked on this object. We create a new instance without deep-copying the invoker object which may contain non-serializable components.

Parameters:

Name	Type	Description	Default
memo	dict	Memoization dictionary for deepcopy process	required

Returns:

Name	Type	Description
LightRAGLMInvokerAdapter	LightRAGLMInvokerAdapter	A new instance with the same invoker reference

LlamaIndexEMInvokerAdapter(em_invoker, embed_batch_size=DEFAULT_EMBED_BATCH_SIZE, **kwargs)

Bases: BaseEmbedding

Minimal EMInvoker adapter for the LlamaIndex BaseEmbedding interface.

This adapter wraps a BaseEMInvoker instance to provide compatibility with LlamaIndex's BaseEmbedding interface. Embeddings from the underlying invoker are returned directly without any conversion, assuming they are already in the correct format (list of floats).

The adapter provides both synchronous and asynchronous methods for: - Query embeddings: Single text embedding for search queries - Text embeddings: Single or batch text embedding for documents

Attributes:

Name	Type	Description
em_invoker	BaseEMInvoker	The underlying EM invoker instance.
model_name	str	The name of the embedding model (inherited from invoker).
embed_batch_size	int	The batch size for batch embedding operations.

Note

Sync methods (get*) use asyncio.run internally to call async methods. The implementation uses nest_asyncio to handle nested event loops if needed.

Initialize the LlamaIndexEMInvokerAdapter.

Parameters:

Name	Type	Description	Default
em_invoker	BaseEMInvoker	The EM invoker to wrap.	required
embed_batch_size	int	The batch size for embedding operations. Defaults to DEFAULT_EMBED_BATCH_SIZE from LlamaIndex.	DEFAULT_EMBED_BATCH_SIZE
**kwargs	Any	Additional keyword arguments passed to BaseEmbedding (e.g., callback_manager).	{}

class_name() classmethod

Get the class name (implements BaseEmbedding.class_name).

This is used by LlamaIndex for serialization and debugging.

Returns:

Name	Type	Description
str	str	The class name "LlamaIndexEMInvokerAdapter".

LlamaIndexLMInvokerAdapter(lm_invoker, **kwargs)

Bases: LLM

Minimal LMInvoker adapter for the LlamaIndex LLM interface.

This adapter wraps a BaseLMInvoker instance to provide compatibility with LlamaIndex's LLM interface. It handles conversion between GLLM message formats and LlamaIndex ChatMessage formats.

Only chat functionality is implemented. Completion and streaming methods raise NotImplementedError to keep the implementation minimal.

Attributes:

Name	Type	Description
lm_invoker	BaseLMInvoker	The underlying LM invoker instance.

Note

Message roles are converted using the ROLE_MAPPING constant, which maps all LlamaIndex message roles (SYSTEM, DEVELOPER, USER, ASSISTANT, TOOL, FUNCTION, CHATBOT, MODEL) to GLLM MessageRole values.

Initialize the LlamaIndexLMInvokerAdapter.

Parameters:

Name	Type	Description	Default
lm_invoker	BaseLMInvoker	The LM invoker to wrap.	required
**kwargs	Any	Additional keyword arguments.	{}

metadata property

Get metadata about the language model.

Returns:

Name	Type	Description
LLMMetadata	LLMMetadata	Metadata containing model information.

achat(messages, **kwargs) async

Asynchronous chat endpoint (implements LlamaIndex LLM.achat).

Converts LlamaIndex ChatMessage objects to GLLM Message format, invokes the underlying LM invoker asynchronously, and converts the response back to ChatResponse.

Parameters:

Name	Type	Description	Default
messages	Sequence[ChatMessage]	The chat messages in LlamaIndex format.	required
**kwargs	Any	Additional keyword arguments. Supports: - hyperparameters (dict, optional): Model hyperparameters like temperature, max_tokens, etc.	{}

Returns:

Name	Type	Description
ChatResponse	ChatResponse	The chat response in LlamaIndex format with message content, role, and optional metadata (token usage, finish details).

acomplete(prompt, formatted=False, **kwargs) async

Asynchronous completion endpoint.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt string.	required
formatted	bool	Whether the prompt is already formatted. Defaults to False.	False
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Name	Type	Description
CompletionResponse	CompletionResponse	The completion response.

Raises:

Type	Description
NotImplementedError	Always raises this exception.

astream_chat(messages, **kwargs)

Asynchronous streaming chat endpoint.

Parameters:

Name	Type	Description	Default
messages	Sequence[ChatMessage]	The chat messages.	required
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Name	Type	Description
ChatResponse	AsyncGenerator[ChatResponse, None]	Streaming chat responses.

Raises:

Type	Description
NotImplementedError	Always raises this exception.

astream_complete(prompt, formatted=False, **kwargs)

Asynchronous streaming completion endpoint.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt string.	required
formatted	bool	Whether the prompt is already formatted. Defaults to False.	False
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Name	Type	Description
CompletionResponse	AsyncGenerator[CompletionResponse, None]	Streaming completion responses.

Raises:

Type	Description
NotImplementedError	Always raises this exception.

chat(messages, **kwargs)

Synchronous chat endpoint (implements LlamaIndex LLM.chat).

This is a synchronous wrapper around the async achat() method. It handles both scenarios: when called from within an event loop and when called from synchronous code.

Converts LlamaIndex ChatMessage objects to GLLM Message format, invokes the underlying LM invoker, and converts the response back to ChatResponse.

Parameters:

Name	Type	Description	Default
messages	Sequence[ChatMessage]	The chat messages in LlamaIndex format.	required
**kwargs	Any	Additional keyword arguments. Supports: - hyperparameters (dict, optional): Model hyperparameters like temperature, max_tokens, etc.	{}

Returns:

Name	Type	Description
ChatResponse	ChatResponse	The chat response in LlamaIndex format with message content, role, and optional metadata (token usage, finish details).

class_name() classmethod

Get the class name (implements LLM.class_name).

This is used by LlamaIndex for serialization and debugging.

Returns:

Name	Type	Description
str	str	The class name "LlamaIndexLMInvokerAdapter".

complete(prompt, formatted=False, **kwargs)

Synchronous completion endpoint.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt string.	required
formatted	bool	Whether the prompt is already formatted. Defaults to False.	False
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Name	Type	Description
CompletionResponse	CompletionResponse	The completion response.

Raises:

Type	Description
NotImplementedError	Always raises this exception.

stream_chat(messages, **kwargs)

Streaming chat endpoint.

Parameters:

Name	Type	Description	Default
messages	Sequence[ChatMessage]	The chat messages.	required
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Name	Type	Description
ChatResponse	AsyncGenerator[ChatResponse, None]	Streaming chat responses.

Raises:

Type	Description
NotImplementedError	Always raises this exception.

stream_complete(prompt, formatted=False, **kwargs)

Streaming completion endpoint.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt string.	required
formatted	bool	Whether the prompt is already formatted. Defaults to False.	False
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Name	Type	Description
CompletionResponse	AsyncGenerator[CompletionResponse, None]	Streaming completion responses.

Raises:

Type	Description
NotImplementedError	Always raises this exception.

NoOpGraphitiReranker

Bases: CrossEncoderClient

No-op cross encoder that returns passages in original order.

This cross encoder implementation does not perform any reranking. It simply returns the passages in their original order with a score of 0.0. This is useful when you want to disable reranking functionality.

rank(query, passages) async

Return passages in original order without reranking.

Parameters:

Name	Type	Description	Default
query	str	The query string (unused).	required
passages	list[str]	A list of passages to rank.	required

Returns:

Type	Description
list[tuple[str, float]]	list[tuple[str, float]]: A list of tuples containing the passage and a score of 0.0, in the original order.