cohere_tracer

NonStreamedChatResponse

Bases: NonStreamedChatResponse

Wrapper of cohere.types.non_streamed_chat_response.NonStreamedChatResponse with ImpactsOutput

StreamEndStreamedChatResponse

Bases: StreamEndStreamedChatResponse

Wrapper of cohere.types.streamed_chat_response.StreamEndStreamedChatResponse with ImpactsOutput

CohereInstrumentor()

Instrumentor initialized by EcoLogits to automatically wrap all Cohere calls

Source code in ecologits/tracers/cohere_tracer.py

def __init__(self) -> None:
    self.wrapped_methods = [
        {
            "module": "cohere.base_client",
            "name": "BaseCohere.chat",
            "wrapper": cohere_chat_wrapper,
        },
        {
            "module": "cohere.base_client",
            "name": "AsyncBaseCohere.chat",
            "wrapper": cohere_async_chat_wrapper,
        },
        {
            "module": "cohere.base_client",
            "name": "BaseCohere.chat_stream",
            "wrapper": cohere_stream_chat_wrapper,
        },
        {
            "module": "cohere.base_client",
            "name": "AsyncBaseCohere.chat_stream",
            "wrapper": cohere_async_stream_chat_wrapper,
        },
    ]

cohere_chat_wrapper(wrapped, instance, args, kwargs)

Function that wraps a Cohere answer with computed impacts

Parameters:

Name	Type	Description	Default
wrapped	Callable	Callable that returns the LLM response	required
instance	Client	Never used - for compatibility with wrapt	required
args	Any	Arguments of the callable	required
kwargs	Any	Keyword arguments of the callable	required

Returns:

Type	Description
NonStreamedChatResponse	A wrapped NonStreamedChatResponse with impacts

Source code in ecologits/tracers/cohere_tracer.py

def cohere_chat_wrapper(
    wrapped: Callable, instance: Client, args: Any, kwargs: Any  # noqa: ARG001
) -> NonStreamedChatResponse:
    """
    Function that wraps a Cohere answer with computed impacts

    Args:
        wrapped: Callable that returns the LLM response
        instance: Never used - for compatibility with `wrapt`
        args: Arguments of the callable
        kwargs: Keyword arguments of the callable

    Returns:
        A wrapped `NonStreamedChatResponse` with impacts
    """
    timer_start = time.perf_counter()
    response = wrapped(*args, **kwargs)
    request_latency = time.perf_counter() - timer_start
    output_tokens = response.meta.tokens.output_tokens
    model_name = kwargs.get("model", "command-r")
    impacts = llm_impacts(
        provider=PROVIDER,
        model_name=model_name,
        output_token_count=output_tokens,
        request_latency=request_latency,
        electricity_mix_zone=EcoLogits.config.electricity_mix_zone
    )
    return NonStreamedChatResponse(**response.dict(), impacts=impacts)

cohere_async_chat_wrapper(wrapped, instance, args, kwargs) async

Function that wraps a Cohere answer with computed impacts in async mode

Parameters:

Name	Type	Description	Default
wrapped	Callable	Async callable that returns the LLM response	required
instance	AsyncClient	Never used - for compatibility with wrapt	required
args	Any	Arguments of the callable	required
kwargs	Any	Keyword arguments of the callable	required

Returns:

Type	Description
NonStreamedChatResponse	A wrapped NonStreamedChatResponse with impacts

Source code in ecologits/tracers/cohere_tracer.py

async def cohere_async_chat_wrapper(
    wrapped: Callable, instance: AsyncClient, args: Any, kwargs: Any    # noqa: ARG001
) -> NonStreamedChatResponse:
    """
    Function that wraps a Cohere answer with computed impacts in async mode

    Args:
        wrapped: Async callable that returns the LLM response
        instance: Never used - for compatibility with `wrapt`
        args: Arguments of the callable
        kwargs: Keyword arguments of the callable

    Returns:
        A wrapped `NonStreamedChatResponse` with impacts
    """
    timer_start = time.perf_counter()
    response = await wrapped(*args, **kwargs)
    request_latency = time.perf_counter() - timer_start
    output_tokens = response.meta.tokens.output_tokens
    model_name = kwargs.get("model", "command-r")
    impacts = llm_impacts(
        provider=PROVIDER,
        model_name=model_name,
        output_token_count=output_tokens,
        request_latency=request_latency,
        electricity_mix_zone=EcoLogits.config.electricity_mix_zone
    )
    return NonStreamedChatResponse(**response.dict(), impacts=impacts)

cohere_stream_chat_wrapper(wrapped, instance, args, kwargs)

Function that wraps a Cohere answer with computed impacts in streaming mode

Parameters:

Name	Type	Description	Default
wrapped	Callable	Callable that returns the LLM response	required
instance	Client	Never used - for compatibility with wrapt	required
args	Any	Arguments of the callable	required
kwargs	Any	Keyword arguments of the callable	required

Returns:

Type	Description
Iterator[StreamedChatResponse]	A wrapped Iterator[StreamedChatResponse] with impacts

Source code in ecologits/tracers/cohere_tracer.py

def cohere_stream_chat_wrapper(
    wrapped: Callable, instance: Client, args: Any, kwargs: Any # noqa: ARG001
) -> Iterator[StreamedChatResponse]:
    """
    Function that wraps a Cohere answer with computed impacts in streaming mode

    Args:
        wrapped: Callable that returns the LLM response
        instance: Never used - for compatibility with `wrapt`
        args: Arguments of the callable
        kwargs: Keyword arguments of the callable

    Returns:
        A wrapped `Iterator[StreamedChatResponse]` with impacts
    """

    model_name = kwargs.get("model", "command-r")
    timer_start = time.perf_counter()
    stream = wrapped(*args, **kwargs)
    for event in stream:
        if event.event_type == "stream-end":
            request_latency = time.perf_counter() - timer_start
            output_tokens = event.response.meta.tokens.output_tokens
            impacts = llm_impacts(
                provider=PROVIDER,
                model_name=model_name,
                output_token_count=output_tokens,
                request_latency=request_latency,
                electricity_mix_zone=EcoLogits.config.electricity_mix_zone
            )
            yield StreamEndStreamedChatResponse(**event.dict(), impacts=impacts)
        else:
            yield event

cohere_async_stream_chat_wrapper(wrapped, instance, args, kwargs) async

Function that wraps a Cohere answer with computed impacts in streaming and async mode

Parameters:

Name	Type	Description	Default
wrapped	Callable	Callable that returns the LLM response	required
instance	AsyncClient	Never used - for compatibility with wrapt	required
args	Any	Arguments of the callable	required
kwargs	Any	Keyword arguments of the callable	required

Returns:

Type	Description
AsyncIterator[StreamedChatResponse]	A wrapped AsyncIterator[StreamedChatResponse] with impacts

Source code in ecologits/tracers/cohere_tracer.py

async def cohere_async_stream_chat_wrapper(
    wrapped: Callable, instance: AsyncClient, args: Any, kwargs: Any # noqa: ARG001
) -> AsyncIterator[StreamedChatResponse]:
    """
    Function that wraps a Cohere answer with computed impacts in streaming and async mode

    Args:
        wrapped: Callable that returns the LLM response
        instance: Never used - for compatibility with `wrapt`
        args: Arguments of the callable
        kwargs: Keyword arguments of the callable

    Returns:
        A wrapped `AsyncIterator[StreamedChatResponse]` with impacts
    """

    model_name = kwargs.get("model", "command-r")
    timer_start = time.perf_counter()
    stream = wrapped(*args, **kwargs)
    async for event in stream:
        if event.event_type == "stream-end":
            request_latency = time.perf_counter() - timer_start
            output_tokens = event.response.meta.tokens.output_tokens
            impacts = llm_impacts(
                provider=PROVIDER,
                model_name=model_name,
                output_token_count=output_tokens,
                request_latency=request_latency,
                electricity_mix_zone=EcoLogits.config.electricity_mix_zone
            )
            yield StreamEndStreamedChatResponse(**event.dict(), impacts=impacts)
        else:
            yield event