Tutorial

Usage Example

Sample Code

python

from elsai_arms.elsai_arms import ElsaiARMS
from elsai_model.azure_openai import AzureOpenAIConnector
import easyocr

arms = ElsaiARMS("Project_Name")

try:
    connector = AzureOpenAIConnector()
    llm = connector.connect_azure_open_ai("gpt-4o-mini")

    arms.info("Script started")

    @arms.monitor_llm_call
    def get_response(prompt: str):
        return llm.invoke(prompt)

    @arms.monitor_ocr_call(model_name="EasyOCR")
    def run_easyocr(image_path):
        reader = easyocr.Reader(['en'])
        return reader.readtext(image_path, detail=1)

    @arms.monitor_embedding_call
    def get_embedding(text):
        # Code for generating embeddings
        embedding = ["Generated Embeddings"]
        return embedding

    @arms.monitor_rag_call
    def retrieve_documents(query, docs, top_k=2):
        # Code for RAG
        results = ["RAG Results"]
        return results[:top_k]

    llm_result = get_response("Tell me a joke")
    ocr_result = run_easyocr("path_to_image.png")

    documents = ["Load Documents"]

    query = "Enter query"
    embedding = get_embedding(query)
    retrieved_docs = retrieve_documents(query, documents)

    arms.log_custom_metric("Number of response", 1)
    arms.info("Run completed")
    arms.warning("This is a sample code")

    arms.end_run()

    arms.flush()

    print(arms.export())

except Exception as e:
    arms.error(f"An error occurred: {e}")

Steps

Initialize elsai ARMS in the application

python

arms = ElsaiARMS('Project_Name')

The Project Manager checks if it's a new project or an existing project

Project Manager creates a new project, if new, otherwise loads existing project

A new run is initiated under the project

To automatically capture LLM-specific metrics

python

@arms.monitor_llm_call

To monitor async streaming LLM calls

python

@arms.monitor_llm_astream
async def run_astream(prompt, llm):
    return llm.astream_events([HumanMessage(content=prompt)])

To capture OCR metrics

python

@arms.monitor_ocr_call

To monitor RAG performance

python

@arms.monitor_rag_call

To monitor embedding metrics

python

@arms.monitor_embedding_call

To monitor LangChain agents and graphs

python

graph.invoke(
    {"messages": messages},
    config={"callbacks": [arms.langchain_callback]}
)

To monitor elsai Agents (on-prem)

python

from elsai.agent import AgentConfig

agent = Agent(
    model=model,
    tools=[...],
    config=AgentConfig(
        name="manager",
        hooks=[arms.elsai_agents_hook],
    ),
)
result = await agent.invoke_async("...")

Requires pip install 'elsai-arms[elsai-agents]'. See elsai Agents monitoring.

To log custom metrics

python

arms.log_custom_metric('Metric Name', metric_value)

To log operation

python

arms.info('Log Operation')

To log warning

python

arms.warning('Log Warning')

To log errors

python

arms.error('Log Error')

Once metrics are logged, the exporter module is used to export logs in JSON format. To export data

python

arms.export()

Once the run is done, the project manager saves the run details and the project details are stored in MongoDB. To end run

python

arms.end_run()

Async buffered flush: run data is buffered in memory and sent to the backend in the background (by time, event count, or buffer size). Optional: use manual flush before exit or when you need data persisted immediately.

python

# Automatic flush: time, event count, or buffer size
arms.configure_flush(
    interval_seconds=15,   # time-based flush
    event_count=5,         # event-count flush
    size_bytes=2_000_000   # size-based flush (bytes)
)
# Manual flush (optional): send buffered data immediately
arms.flush()

Tutorial ​

Usage Example ​

Sample Code ​

Steps ​

Tutorial

Usage Example

Sample Code

Steps