llm (Large Language Model)

Caching Wrapper

class service.llm.base.BASE_LLM_CACHE[source]

Bases: object

A base class for implementing caching of queries and responses for a Language Model (LLM). This class is intended to be inherited and implemented by a subclass.

check_cache(query)[source]

Checks if a cached response exists for the provided query.

Parameters:: query – The query to check in the cache.
Returns:: The cached response if available, otherwise a constant (NO_CACHE_YET). Also updates the internal usage cost for cached responses.

clear_usage()[source]: Clears the stored usage statistics.

print_usage()[source]: Prints the cumulative token usage statistics.

write_cache(query, response, usage)[source]

Writes the provided query, response, and usage information to the cache.

Parameters:

query – The query to be cached.
response – The response to be cached.
usage – The token usage information.

Mock API

class service.llm.mock.MockLLMService[source]

Bases: object

classmethod check_llm_job_done(*args, **kwargs)[source]

classmethod get_llm_job_response(*args, **kwargs)[source]

classmethod trigger(*args, **kwargs)[source]

OpenAI API

class service.llm.openai.OPENAI(api_key, cache_collection=None, **kwargs)[source]

Bases: BASE_LLM_CACHE

A class to interact with OpenAI’s language model while using a cache mechanism to optimize requests. Inherits from BASE_LLM_CACHE.

class service.llm.openai.OPENAI_SERVICE[source]

Bases: object

A service class for managing OPENAI LLM requests through a queue mechanism using MongoDB and RabbitMQ.

collection = Collection(Database(MongoClient(host=['localhost:27017'], document_class=dict, tz_aware=False, connect=True), 'llm'), 'openai')

static get_response(job_id)[source]

Retrieves the response of a job with the given ID.

Parameters:: job_id (ObjectId) – The ID of the job to retrieve the response for.
Returns:: The response of the job, or None if the job is not found or has not completed.
Return type:: str

static get_response_sync(job_id, timeout=300)[source]

Retrieves the response of a job with the given ID synchronously.

Parameters:

job_id (ObjectId) – The ID of the job to retrieve the response for.
timeout (int, optional) – The maximum time to wait for the job to complete.

Returns:

The response of the job, or None if the job is not found or has not completed within the timeout.

Return type:

str

static launch_worker()[source]: Launches a worker to process jobs from the RabbitMQ queue. The worker interacts with the LLM and stores the response back in MongoDB.

logger = <Logger service.llm.openai (INFO)>

queue_name = 'llm-openai'

static trigger(parent_service, parent_job_id=None, use_cache=False, **query)[source]

Creates and triggers a new job for an LLM request.

Parameters:

caller_service (str) – The service initiating the request.
use_cache (bool, optional) – Whether to use cached responses if available.
**query – The query to send to the LLM.

Returns:

The job ID of the triggered request.

Return type:

str

service.llm.openai.now(tz=None)

Returns new datetime object representing current time local to tz.

tz
Timezone object.

If no tz is specified, uses local timezone.

ZhipuAI API

class service.llm.zhipuai.ZHIPUAI(api_key, cache_collection=None, **kwargs)[source]

Bases: BASE_LLM_CACHE

call_model(query, use_cache)[source]

Calls the ZhipuAI model with the given query.

Parameters:

query (dict) – The query to send to the OpenAI model.
use_cache (bool) – Whether to use cached responses if available.

Returns:

The response from the model, either from cache or a new request.

Return type:

str

class service.llm.zhipuai.ZHIPUAI_SERVICE[source]

Bases: object

A service class for managing ZhipuAI LLM requests through a queue mechanism using MongoDB and RabbitMQ.

classmethod check_llm_job_done(job_id)[source]

Checks if a job has been completed.

Parameters:: job_id (str) – The ID of the job to check.
Returns:: Whether the job has been completed.
Return type:: bool

collection = Collection(Database(MongoClient(host=['localhost:27017'], document_class=dict, tz_aware=False, connect=True), 'llm'), 'zhipu')

classmethod get_llm_job_response(job_id)[source]

Gets the response of a completed job.

Parameters:: job_id (str) – The ID of the job to check.
Returns:: The response of the job.
Return type:: str

classmethod launch_worker()[source]: Launches a worker to process jobs from the RabbitMQ queue. The worker interacts with the LLM and stores the response back in MongoDB.

queue_name = 'llm-zhipu'

classmethod trigger(query, caller_service, use_cache=False)[source]

Creates and triggers a new job for an LLM request.

Parameters:

query (dict) – The query to send to the LLM.
caller_service (str) – The service initiating the request.
use_cache (bool) – Whether to use cached responses if available.

Returns:

The job ID of the triggered request.

Return type:

str

service.llm.zhipuai.now(tz=None)

Returns new datetime object representing current time local to tz.

tz
Timezone object.

If no tz is specified, uses local timezone.