MyScale Telemetry：开源版 LangSmith，增强 LLM 应用程序的可观测性

CREATE TABLE your_database_name.your_table_name
(
    `TraceId` String CODEC(ZSTD(1)),
    `SpanId` String CODEC(ZSTD(1)),
    `ParentSpanId` String CODEC(ZSTD(1)),
    `StartTime` DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    `EndTime` DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    `Duration` Int64 CODEC(ZSTD(1)),
    `SpanName` LowCardinality(String) CODEC(ZSTD(1)),
    `SpanKind` LowCardinality(String) CODEC(ZSTD(1)),
    `ServiceName` LowCardinality(String) CODEC(ZSTD(1)),
    `SpanAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    `ResourceAttributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    `StatusCode` LowCardinality(String) CODEC(ZSTD(1)),
    `StatusMessage` String CODEC(ZSTD(1)),
    INDEX idx_trace_id TraceId TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_attr_key mapKeys(SpanAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_attr_value mapValues(SpanAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_duration Duration TYPE minmax GRANULARITY 1
)
ENGINE = MergeTree()
PARTITION BY toDate(StartTime)
ORDER BY (SpanName, toUnixTimestamp(StartTime), TraceId);

git clone https://github.com/myscale/myscale-telemetry.git
cd myscale-telemetry/deploy/
docker-compose up -d

docker-compose up --scale myscaledb=0 -d

pip install myscale-telemetry langchain_openai ragas

import os
from myscale_telemetry.handler import MyScaleCallbackHandler
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import MyScale
from langchain_community.vectorstores.myscale import MyScaleSettings
from langchain_core.runnables import RunnableConfig
 
# set up the environment variables for OpenAI and MyScale Cloud/MyScaleDB:
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_KEY"
os.environ["MYSCALE_HOST"] = "YOUR_MYSCALE_HOST"
os.environ["MYSCALE_PORT"] = "YOUR_MYSCALE_HOST"
os.environ["MYSCALE_USERNAME"] = "YOUR_MYSCALE_USERNAME"
os.environ["MYSCALE_PASSWORD"] = "YOUR_MYSCALE_PASSWORD"
 
# create a vector store and retriever using MyScale and OpenAI embeddings:
texts = [
  "Harrison worked at Kensho.",
  "Alice is a software engineer.",
  "Bob enjoys hiking on weekends.",
  "Claire is studying data science.",
  "David works at a tech startup.",
  "Eva loves playing the piano.",
  "Frank is a graphic designer.",
  "Grace is an artificial intelligence researcher.",
  "Henry is a freelance writer.",
  "Isabel is learning machine learning."
]
 
myscale_settings = MyScaleSettings()
myscale_settings.index_type = 'SCANN'
vectorstore = MyScale.from_texts(texts, embedding=OpenAIEmbeddings(), config=myscale_settings)
retriever = vectorstore.as_retriever()
 
# set up the llm and prompt template:
model = ChatOpenAI()
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
 
# create the chain
chain = (
    {
    "context": itemgetter("question") | retriever,
    "question": itemgetter("question"),
    }
    | prompt
    | model
    | StrOutputParser()
)
 
# integrate MyScaleCallbackHandler to capture trace data during the chain execution:
chain.invoke({"question": "where did harrison work"}, config=RunnableConfig(
    callbacks=[
        MyScaleCallbackHandler()
    ]
))

import os
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_utilization
from clickhouse_connect import get_client
 
 
def score_with_ragas(query, chunks, answer):
  test_dataset = Dataset.from_dict({"question": [query], "contexts": [chunks], "answer": [answer]})
  result = evaluate(test_dataset, metrics=[faithfulness, answer_relevancy, context_utilization])
  return result
 
 
def evaluate_trace(question, topk, client, database_name, table_name):
  trace_id, answer = client.query(
    f"SELECT TraceId, SpanAttributes['output'] as Answer FROM {database_name}.{table_name} WHERE SpanAttributes['question'] = '{question}' AND ParentSpanId = ''"
  ).result_rows[0]
 
  span_dict = client.query(
    f"SELECT SpanAttributes FROM {database_name}.{table_name} WHERE TraceId = '{trace_id}' AND SpanKind = 'retriever'"
  ).result_rows[0][0]
 
  contexts = [span_dict.get(f"documents.{i}.content") for i in range(topk)]
 
  print(score_with_ragas(question, contexts, answer))
 
 
test_question = "where did harrison work"
client = get_client(
  host=os.getenv("MYSCALE_HOST"),
  port=int(os.getenv("MYSCALE_PORT")),
  username=os.getenv("MYSCALE_USERNAME"),
  password=os.getenv("MYSCALE_PASSWORD"),
)
evaluate_trace(test_question, 4, client, "otel", "otel_traces")