Loading...
Loading...
Compare original and translation side by side
pip install langsmithpip install langsmithundefinedundefinedfrom langsmith import traceable
from openai import OpenAI
client = OpenAI()
@traceable
def generate_response(prompt: str) -> str:
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.contentfrom langsmith import traceable
from openai import OpenAI
client = OpenAI()
@traceable
def generate_response(prompt: str) -> str:
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.contentundefinedundefinedfrom langsmith.wrappers import wrap_openai
from openai import OpenAIfrom langsmith.wrappers import wrap_openai
from openai import OpenAIundefinedundefinedfrom langsmith import traceable
@traceable(run_type="chain")
def process_query(query: str) -> str:
# Parent run
context = retrieve_context(query) # Child run
response = generate_answer(query, context) # Child run
return response
@traceable(run_type="retriever")
def retrieve_context(query: str) -> list:
return vector_store.search(query)
@traceable(run_type="llm")
def generate_answer(query: str, context: list) -> str:
return llm.invoke(f"Context: {context}\n\nQuestion: {query}")from langsmith import traceable
@traceable(run_type="chain")
def process_query(query: str) -> str:
# 父运行
context = retrieve_context(query) # 子运行
response = generate_answer(query, context) # 子运行
return response
@traceable(run_type="retriever")
def retrieve_context(query: str) -> list:
return vector_store.search(query)
@traceable(run_type="llm")
def generate_answer(query: str, context: list) -> str:
return llm.invoke(f"Context: {context}\n\nQuestion: {query}")import os
os.environ["LANGSMITH_PROJECT"] = "my-project"import os
os.environ["LANGSMITH_PROJECT"] = "my-project"undefinedundefinedfrom langsmith import Client
client = Client()from langsmith import Client
client = Client()undefinedundefinedfrom langsmith import Client
client = Client()from langsmith import Client
client = Client()undefinedundefinedfrom langsmith import evaluate
def my_model(inputs: dict) -> dict:
# Your model logic
return {"answer": generate_answer(inputs["question"])}
def correctness_evaluator(run, example):
prediction = run.outputs["answer"]
reference = example.outputs["answer"]
score = 1.0 if reference.lower() in prediction.lower() else 0.0
return {"key": "correctness", "score": score}
results = evaluate(
my_model,
data="qa-test-set",
evaluators=[correctness_evaluator],
experiment_prefix="v1"
)
print(f"Average score: {results.aggregate_metrics['correctness']}")from langsmith import evaluate
def my_model(inputs: dict) -> dict:
# 你的模型逻辑
return {"answer": generate_answer(inputs["question"])}
def correctness_evaluator(run, example):
prediction = run.outputs["answer"]
reference = example.outputs["answer"]
score = 1.0 if reference.lower() in prediction.lower() else 0.0
return {"key": "correctness", "score": score}
results = evaluate(
my_model,
data="qa-test-set",
evaluators=[correctness_evaluator],
experiment_prefix="v1"
)
print(f"Average score: {results.aggregate_metrics['correctness']}")from langsmith.evaluation import LangChainStringEvaluatorfrom langsmith.evaluation import LangChainStringEvaluatorundefinedundefinedfrom langsmith import tracing_context
with tracing_context(
project_name="experiment-1",
tags=["production", "v2"],
metadata={"version": "2.0"}
):
# All traceable calls inherit context
result = my_function()from langsmith import tracing_context
with tracing_context(
project_name="experiment-1",
tags=["production", "v2"],
metadata={"version": "2.0"}
):
# 所有可追踪调用都会继承上下文配置
result = my_function()from langsmith import trace
with trace(
name="custom_operation",
run_type="tool",
inputs={"query": "test"}
) as run:
result = do_something()
run.end(outputs={"result": result})from langsmith import trace
with trace(
name="custom_operation",
run_type="tool",
inputs={"query": "test"}
) as run:
result = do_something()
run.end(outputs={"result": result})def sanitize_inputs(inputs: dict) -> dict:
if "password" in inputs:
inputs["password"] = "***"
return inputs
@traceable(process_inputs=sanitize_inputs)
def login(username: str, password: str):
return authenticate(username, password)def sanitize_inputs(inputs: dict) -> dict:
if "password" in inputs:
inputs["password"] = "***"
return inputs
@traceable(process_inputs=sanitize_inputs)
def login(username: str, password: str):
return authenticate(username, password)import os
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1" # 10% samplingimport os
os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = "0.1" # 10%采样率from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplateundefinedundefinedfrom langsmith import Client
client = Client()from langsmith import Client
client = Client()undefinedundefinedfrom langsmith import AsyncClient
async def main():
client = AsyncClient()
runs = []
async for run in client.list_runs(project_name="my-project"):
runs.append(run)
return runsfrom langsmith import AsyncClient
async def main():
client = AsyncClient()
runs = []
async for run in client.list_runs(project_name="my-project"):
runs.append(run)
return runsfrom langsmith import Client
client = Client()from langsmith import Client
client = Client()undefinedundefinedfrom langsmith import test
@test
def test_qa_accuracy():
result = my_qa_function("What is Python?")
assert "programming" in result.lower()from langsmith import test
@test
def test_qa_accuracy():
result = my_qa_function("What is Python?")
assert "programming" in result.lower()from langsmith import evaluate
def run_evaluation():
results = evaluate(
my_model,
data="regression-test-set",
evaluators=[accuracy_evaluator]
)
# Fail CI if accuracy drops
assert results.aggregate_metrics["accuracy"] >= 0.9, \
f"Accuracy {results.aggregate_metrics['accuracy']} below threshold"from langsmith import evaluate
def run_evaluation():
results = evaluate(
my_model,
data="regression-test-set",
evaluators=[accuracy_evaluator]
)
# 如果准确率下降则CI失败
assert results.aggregate_metrics["accuracy"] >= 0.9, \
f"Accuracy {results.aggregate_metrics['accuracy']} below threshold"import osimport os
**High latency from tracing:**
```python
**链路追踪导致延迟过高:**
```python
**Large payloads:**
```python
**负载过大:**
```pythonundefinedundefined