快速开始

# 查看 Python 版本   
python --version 
 
# 回显信息显示已安装 3.8.2 版本 
Python 3.8.2

pip install cozeloop

import os
import cozeloop

if __name__ == "__main__":
    # 设置相关环境变量
    os.environ["COZELOOP_WORKSPACE_ID"] = "your workspace id"
    os.environ["COZELOOP_API_TOKEN"] = "your token"

    # 直接调用api
    span = cozeloop.start_span("first_span", "custom")
    span.finish()

    # 程序退出前，需要调用Close方法，否则可能造成trace数据上报丢失。Close后无法再执行任何操作。
    cozeloop.close()

import os
import cozeloop

if __name__ == "__main__":
    # 设置相关环境变量
    os.environ["COZELOOP_WORKSPACE_ID"] = "your workspace id"
    os.environ["COZELOOP_API_TOKEN"] = "your token"
    
    # 创建client
    client = cozeloop.new_client()

    # 通过client调用api
    span = client.start_span("first_span", "custom")
    span.finish()

    # 程序退出前，需要调用Close方法，否则可能造成trace数据上报丢失。Close后无法再执行任何操作。
    client.close() 

import os

import cozeloop
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

from cozeloop.integration.langchain.trace_callback import LoopTracer


def do_lcel_demo():
    # 配置CozeLoop环境变量
    os.environ['COZELOOP_API_TOKEN'] = '{your_token}'
    os.environ['COZELOOP_WORKSPACE_ID'] = '{your_workspace_id}'

    # 创建cozeloop client
    client = cozeloop.new_client()
    # 注册callback
    trace_callback_handler = LoopTracer.get_callback_handler(client)

    llm_model = ChatOpenAI()
    lcel_sequence = llm_model | StrOutputParser()
    output = lcel_sequence.invoke(
        input='用你所学的技巧，帮我生成几个有意思的问题',
        config=RunnableConfig(callbacks=[trace_callback_handler])
    )
    print(output)
    
    # 程序退出前，需要调用Close方法，否则可能造成trace数据上报丢失。Close后无法再执行任何操作。
    client.close()


if __name__ == "__main__":
    do_lcel_demo()

import logging
import os
import time

from cozeloop import new_client, flush, get_span_from_context
from cozeloop.decorator import observe
from cozeloop.logger import set_log_level

logger = logging.getLogger(__name__)

ERR_CODE_LLM_CALL = 600789111

class LLMRunner:
    def __init__(self):
        pass

    @observe(
        name="llmCall",
        span_type="model",
        tags={"model_provider":'openai', "input_tokens": 232, "output_tokens": 1211, "model_name": "gpt-4-1106-preview"},
    )
    def llm_call(self, input_data):
        """
        Simulate an LLM call and set relevant span tags.
        """
        # span = self.client.start_span("llmCall", "model")
        try:
            # Assuming llm is processing
            # output = ChatOpenAI().invoke(input=input_data)

            # mock resp
            time.sleep(1)
            output = "I'm a robot. I don't have a specific name. You can give me one."
            return output
        except Exception as e:
            raise e
        finally:
            pass

@observe(
    # client=new_client(api_token="**"), # Unless you need a new client, no configuration is required,
                                         # the default client will be used automatically.
    name="root_span",                    # The name of the Span. Defaults to the function name.
    span_type="main_span",               # The span_type of the Span. Defaults to 'custom'.
    tags={"mode": 'simple', "node_id": 6076665},  # Set static custom tag. The Priority is higher than the default tags.
    baggage={"product_id": "123456654321"},  # Set static custom baggage. baggage can cover tag of sample key, and will pass to child span automatically.
)
def do_simple_demo():
    # Dynamic set tag or baggage in runtime
    get_span_from_context().set_baggage({"thread_id": "my-thread-id"})
    get_span_from_context().set_tags({"sale_id": "my-sale-id"})

    # 0. new client
    set_log_level(logging.INFO)
    llm_runner = LLMRunner()

    # assuming call llm
    try:
        # assuming call llm
        llm_runner.llm_call("What's your name?")
    except Exception as e:
        pass


if __name__ == "__main__":
    # Set the following environment variables first (Assuming you are using a PAT token.).
    # os.environ["COZELOOP_WORKSPACE_ID"] = "your workspace id"
    # os.environ["COZELOOP_API_TOKEN"] = "your token"

    do_simple_demo()

    # flush all trace data before server exit.
    flush()

import cozeloop

def handler(input: str) -> str:
    # 开启root span
    span = cozeloop.start_span("{your_span_name}", "{your_span_type}")

    # 设置用户输入
    span.set_input(input)
    # 设置相关业务id，可以使用Baggage能力将这些id全链路透传，减少重复代码，便于在平台上自由检索
    span.set_user_id_baggage("{user_id}")
    span.set_message_id_baggage("{message_id}")
    try:
    # 执行业务逻辑
        output = do_something(input)
        # 设置服务输出
        span.set_output(output)
        # 如果使用流式返回，记录首Token返回的微秒时间戳，SDK会自动计算首Token耗时
        span.set_start_time_first_resp({start_time_first_resp})
    except Exception as e:
        # 设置错误信息
        span.set_status_code(your_error_code)
        span.set_error(e)
    finally:
        # span 上报
        span.finish()

    return output

from typing import Dict, Any, List

import cozeloop
from cozeloop import entities

# 初始化client，打开prompt trace上报开关
client = cozeloop.new_client(prompt_trace=True)

def get_prompt(params: Dict[str, Any]) -> List[entities.Message]:
    # 当version不为空时，将拉取特定版本的prompt
    # 当version为空时，将拉取最新版本的prompt
    prompt = client.get_prompt(
        prompt_key="{promptKey}", 
        version="{version}"
    )

    # 执行模板变量替换
    return client.prompt_format(
        prompt=prompt,
        variables=params
    )

import cozeloop
from cozeloop import entities
from cozeloop.spec import tracespec

def call_llm(prompt: entities.Prompt) -> None:
    # SpanType使用model
    span = cozeloop.start_span("{your_span_name}", tracespec.V_MODEL_SPAN_TYPE)

    # 设置模型输入
    span.set_input(tracespec.ModelInput(
        message=[
            tracespec.ModelMessage(
                role=tracespec.V_ROLE_SYSTEM,
                content="{system_prompt}",
            ),
            tracespec.ModelMessage(
                role=tracespec.V_ROLE_USER,
                content="{user_prompt}",
            )
        ]
    ))

    # 设置模型提供商和模型名称
    span.set_model_provider("{model_provider}")
    span.set_model_name("{model_name}")
    # 关联PromptKey和Version
    span.set_prompt(prompt)
    # 设置Temperature, MaxTokens等其他参数
    span.set_model_call_options(tracespec.ModelCallOption())

    try:
        # 使用任意方式调用大模型
        # ...

        # 设置模型输出
        span.set_output(tracespec.ModelOutput(
            choices=[
                tracespec.ModelChoice(
                    message=tracespec.ModelMessage(
                        role=tracespec.V_ROLE_ASSISTANT,
                        content="{model_response}",
                    )
                )
            ]
        ))

        # 设置InputTokens和OutputTokens，SDK会自动计算TotalTokens
        span.set_input_tokens({input_tokens})
        span.set_output_tokens({output_tokens})

        # 如果使用流式返回，记录首Token返回的微秒时间戳，SDK会自动计算首Token耗时
        span.set_start_time_first_resp({start_time_first_resp})

    except Exception as e:
        span.set_error(e)
        span.set_status_code({status_code})
        raise
    finally:
        span.finish()

import cozeloop

def custom_func() -> None:
    span = cozeloop.start_span("{your_span_name}", "{your_span_type}")

    span.set_input("{your_input}")
    # 设置自定义tags
    span.set_tags({"key": "value"})

    try:
        # 执行业务逻辑
        output = do_something()
        
        span.set_status_code(0)
        span.set_output(output)
    except Exception as e:
        span.set_error(e)
        span.set_status_code({your_error_code})
        raise
    finally:
        span.finish()