Langfuse와 Mistral AI SDK 통합
Langfuse와 Mistral AI SDK 통합 (Mistral AI SDK Integration with Langfuse)
Python에서 Langfuse를 Mistral AI SDK(v1)와 통합하는 단계별 예시를 제공하는 문서예요. Mistral 언어 모델과의 상호작용을 로그·트레이싱해 AI 애플리케이션의 투명성, 디버깅 용이성, 성능 모니터링을 높이는 방법을 배웁니다.
출처: 문서
본문
이 쿡북은 Python에서 Langfuse를 Mistral AI SDK(v1)와 통합하는 단계별 예시를 제공해요. 이 예시들을 따라 Mistral 언어 모델과의 상호작용을 원활하게 로그·트레이싱하는 방법을 배우게 됩니다.
참고: Langfuse는 [LangChain], [LlamaIndex], [LiteLLM], [다른 프레임워크]에도 네이티브 통합되어 있어요. 그중 하나를 사용한다면 Mistral 모델 사용이 바로 계측됩니다.
개요 (Overview)
이 노트북에서는 Langfuse를 Mistral AI SDK와 통합하는 다양한 사용 사례를 살펴볼게요:
- 기본 LLM 호출: 표준 Mistral 모델 상호작용을 Langfuse의 @observe 데코레이터로 감싸 포괄적인 로깅
- 체인 함수 호출: 여러 모델 상호작용이 연결되어 최종 결과를 만드는 복잡한 워크플로우를 관리·관찰
- 비동기 및 스트리밍 지원: Mistral 모델의 비동기·스트리밍 응답에 Langfuse를 사용해 실시간·동시 상호작용을 추적
- 함수 호출: Mistral과 외부 도구 통합을 구현·관찰해 모델이 커스텀 함수와 API와 상호작용하도록 설정
Mistral SDK나 Langfuse의 @observe 데코레이터에 대한 자세한 안내는 [Mistral SDK 저장소]와 [Langfuse Documentation]을 참조하세요.
Langfuse란? (What is Langfuse?)
[Langfuse]는 오픈소스 LLM 엔지니어링 플랫폼이에요. [traces], [evals], [프롬프트 관리] 같은 기능이 포함되어 LLM 앱을 디버깅하고 개선하는 데 도움을 줍니다.
설정 (Setup)
아직 가입하지 않았다면 [Langfuse에 가입]하고, 프로젝트 설정에서 [API 키]를 복사해 환경에 추가하세요.
%pip install mistralai langfuse
import os
# get keys for your project from https://cloud.langfuse.com
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxx"
os.environ["LANGFUSE_HOST"] = "https://cloud.langfuse.com" # 🇪🇺 EU region
# os.environ["LANGFUSE_HOST"] = "https://us.cloud.langfuse.com" # 🇺🇸 US region
# Your Mistral key
os.environ["MISTRAL_API_KEY"] = "xxx"
Mistral API 키를 환경 변수로 설정해요. 아직 없다면 [Mistral 계정에 가입]하고, 무료 체험판이나 요금제에 [구독]한 뒤 [API 키를 생성]하면 돼요.
from mistralai.client import Mistral
# Initialize Mistral client
mistral_client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
예시 (Examples)
1. 완성 (Completions)
[@observe 데코레이터]를 사용해 Mistral AI SDK를 Langfuse와 통합해요. 이는 LLM 상호작용 로깅·트레이싱에 중요해요. @observe(as_type="generation") 데코레이터는 특히 LLM 상호작용을 기록해 입력, 출력, 모델 파라미터를 포착합니다. 결과 mistral_completion 메서드는 프로젝트 전반에서 사용할 수 있어요.
from langfuse.decorators import langfuse_context, observe
# Function to handle Mistral completion calls, wrapped with @observe to log the LLM interaction
@observe(as_type="generation")
def mistral_completion(**kwargs):
# Clone kwargs to avoid modifying the original input
kwargs_clone = kwargs.copy()
# Extract relevant parameters from kwargs
input = kwargs_clone.pop('messages', None)
model = kwargs_clone.pop('model', None)
min_tokens = kwargs_clone.pop('min_tokens', None)
max_tokens = kwargs_clone.pop('max_tokens', None)
temperature = kwargs_clone.pop('temperature', None)
top_p = kwargs_clone.pop('top_p', None)
# Filter and prepare model parameters for logging
model_parameters = {
"maxTokens": max_tokens,
"minTokens": min_tokens,
"temperature": temperature,
"top_p": top_p
}
model_parameters = {k: v for k, v in model_parameters.items() if v is not None}
# Log the input and model parameters before calling the LLM
langfuse_context.update_current_observation(
input=input,
model=model,
model_parameters=model_parameters,
metadata=kwargs_clone,
)
# Call the Mistral model to generate a response
res = mistral_client.chat.complete(**kwargs)
# Log the usage details and output content after the LLM call
langfuse_context.update_current_observation(
usage={
"input": res.usage.prompt_tokens,
"output": res.usage.completion_tokens
},
output=res.choices[0].message.content
)
# Return the model's response object
return res
선택적으로 다른 함수(API 핸들러, 검색 함수 등)도 데코레이션할 수 있어요.
1.1 간단한 예시
다음 예시에서는 최상위 함수 find_best_painter_from에도 데코레이터를 추가했어요. 이 함수는 @observe(as_type="generation")로 데코레이션된 mistral_completion을 호출합니다. 이 계층적 설정은 여러 LLM 호출과 @observe로 데코레이션된 다른 비-LLM 메서드를 포함하는 복잡한 애플리케이션을 추적하는 데 도움을 줍니다.
langfuse_context.update_current_observation이나 langfuse_context.update_current_trace를 사용해 입력, 출력, 모델 파라미터 같은 추가 세부 정보를 트레이스에 추가할 수 있어요.
@observe()
def find_best_painter_from(country="France"):
response = mistral_completion(
model="mistral-small-latest",
max_tokens=1024,
temperature=0.4,
messages=[
{
"content": "Who is the best painter from {country}? Answer in one short sentence.".format(country=country),
"role": "user",
},
]
)
return response.choices[0].message.content
find_best_painter_from()
1.2 체인 완성 (Chained Completions)
이 예시는 @observe 데코레이터로 여러 LLM 호출을 연결하는 방법을 보여줘요. 첫 호출은 지정된 국가에서 최고의 화가를 식별하고, 두 번째 호출은 그 화가의 이름으로 가장 유명한 그림을 찾습니다. 두 상호작용 모두 위에서 만든 mistral_completion 래핑 메서드를 사용하므로 Langfuse에 기록되어 체인 요청 전체에서 완전한 추적이 보장됩니다.
@observe()
def find_best_painting_from(country="France"):
response = mistral_completion(
model="mistral-small-latest",
max_tokens=1024,
temperature=0.1,
messages=[
{
"content": "Who is the best painter from {country}? Only provide the name.".format(country=country),
"role": "user",
},
]
)
painter_name = response.choices[0].message.content
return mistral_completion(
model="mistral-small-latest",
max_tokens=1024,
messages=[
{
"content": "What is the most famous painting of {painter_name}? Answer in one short sentence.".format(painter_name=painter_name),
"role": "user",
},
]
)
find_best_painting_from("Germany")
2. 스트리밍 완성 (Streaming Completions)
다음 예시는 @observe(as_type="generation") 데코레이터로 Mistral 모델의 스트리밍 응답을 처리하는 방법을 보여줘요. Complete 예시와 유사하지만 실시간 스트림 데이터 처리가 포함됩니다.
이전 예시처럼 스트리밍 함수를 @observe 데코레이터로 감싸 입력, 모델 파라미터, 사용량 세부 정보를 포착해요. 또한 함수는 스트림 출력을 점진적으로 처리하며 각 청크가 수신될 때마다 Langfuse 컨텍스트를 업데이트합니다.
# Wrap streaming function with decorator
@observe(as_type="generation")
def stream_mistral_completion(**kwargs):
kwargs_clone = kwargs.copy()
input = kwargs_clone.pop('messages', None)
model = kwargs_clone.pop('model', None)
min_tokens = kwargs_clone.pop('min_tokens', None)
max_tokens = kwargs_clone.pop('max_tokens', None)
temperature = kwargs_clone.pop('temperature', None)
top_p = kwargs_clone.pop('top_p', None)
model_parameters = {
"maxTokens": max_tokens,
"minTokens": min_tokens,
"temperature": temperature,
"top_p": top_p
}
model_parameters = {k: v for k, v in model_parameters.items() if v is not None}
langfuse_context.update_current_observation(
input=input,
model=model,
model_parameters=model_parameters,
metadata=kwargs_clone,
)
res = mistral_client.chat.stream(**kwargs)
final_response = ""
for chunk in res:
content = chunk.data.choices[0].delta.content
final_response += content
yield content
if chunk.data.choices[0].finish_reason == "stop":
langfuse_context.update_current_observation(
usage={
"input": chunk.data.usage.prompt_tokens,
"output": chunk.data.usage.completion_tokens
},
output=final_response
)
break
# Use stream_mistral_completion as you'd usually use the SDK
@observe()
def stream_find_best_five_painter_from(country="France"):
response_chunks = stream_mistral_completion(
model="mistral-small-latest",
max_tokens=1024,
messages=[
{
"content": "Who are the best five painter from {country}? Answer in one short sentence.".format(country=country),
"role": "user",
},
]
)
final_response = ""
for chunk in response_chunks:
final_response += chunk
# You can also do something with each chunk here if needed
print(chunk)
return final_response
stream_find_best_five_painter_from("Spain")
3. 비동기 완성 (Async Completion)
이 예시는 비동기 컨텍스트에서 @observe 데코레이터를 사용하는 방법을 보여줘요. Mistral 모델과 상호작용하는 async 함수를 감싸 요청과 응답이 모두 Langfuse에 기록되도록 합니다. async 함수는 비차단 LLM 호출을 허용해, 동시성이 필요한 애플리케이션에 적합하면서도 상호작용의 완전한 관찰 가능성을 유지합니다.
# Wrap async function with decorator
@observe(as_type="generation")
async def async_mistral_completion(**kwargs):
kwargs_clone = kwargs.copy()
input = kwargs_clone.pop('messages', None)
model = kwargs_clone.pop('model', None)
min_tokens = kwargs_clone.pop('min_tokens', None)
max_tokens = kwargs_clone.pop('max_tokens', None)
temperature = kwargs_clone.pop('temperature', None)
top_p = kwargs_clone.pop('top_p', None)
model_parameters = {
"maxTokens": max_tokens,
"minTokens": min_tokens,
"temperature": temperature,
"top_p": top_p
}
model_parameters = {k: v for k, v in model_parameters.items() if v is not None}
langfuse_context.update_current_observation(
input=input,
model=model,
model_parameters=model_parameters,
metadata=kwargs_clone,
)
res = await mistral_client.chat.complete_async(**kwargs)
langfuse_context.update_current_observation(
usage={
"input": res.usage.prompt_tokens,
"output": res.usage.completion_tokens
},
output=res.choices[0].message.content
)
return res
@observe()
async def async_find_best_musician_from(country="France"):
response = await async_mistral_completion(
model="mistral-small-latest",
max_tokens=1024,
messages=[
{
"content": "Who is the best musician from {country}? Answer in one short sentence.".format(country=country),
"role": "user",
},
]
)
return response
await async_find_best_musician_from("Spain")
4. 비동기 스트리밍 (Async Streaming)
이 예시는 비동기 스트리밍 컨텍스트에서 @observe 데코레이터를 사용하는 방법을 보여줘요. Mistral 모델에서 응답을 스트리밍하는 async 함수를 감싸 각 데이터 청크를 실시간으로 기록합니다.
import asyncio
# Wrap async streaming function with decorator
@observe(as_type="generation")
async def async_stream_mistral_completion(**kwargs):
kwargs_clone = kwargs.copy()
input = kwargs_clone.pop('messages', None)
model = kwargs_clone.pop('model', None)
min_tokens = kwargs_clone.pop('min_tokens', None)
max_tokens = kwargs_clone.pop('max_tokens', None)
temperature = kwargs_clone.pop('temperature', None)
top_p = kwargs_clone.pop('top_p', None)
model_parameters = {
"maxTokens": max_tokens,
"minTokens": min_tokens,
"temperature": temperature,
"top_p": top_p
}
model_parameters = {k: v for k, v in model_parameters.items() if v is not None}
langfuse_context.update_current_observation(
input=input,
model=model,
model_parameters=model_parameters,
metadata=kwargs_clone,
)
res = await mistral_client.chat.stream_async(**kwargs)
final_response = ""
async for chunk in res:
content = chunk.data.choices[0].delta.content
final_response += content
yield content
if chunk.data.choices[0].finish_reason == "stop":
langfuse_context.update_current_observation(
usage={
"input": chunk.data.usage.prompt_tokens,
"output": chunk.data.usage.completion_tokens
},
output=final_response
)
break
@observe()
async def async_stream_find_best_five_musician_from(country="France"):
response_chunks = async_stream_mistral_completion(
model="mistral-small-latest",
max_tokens=1024,
messages=[
{
"content": "Who are the best five musician from {country}? Answer in one short sentence.".format(country=country),
"role": "user",
},
]
)
final_response = ""
async for chunk in response_chunks:
final_response += chunk
# You can also do something with each chunk here if needed
print(chunk)
return final_response
# Run the async function
await async_stream_find_best_five_musician_from("Spain")
5. 도구 호출 (Tool Calling)
이 스니펫은 Mistral의 함수 호출 기능을 소개해요. 트랜잭션 ID를 기준으로 지불 상태와 날짜 같은 특정 데이터를 검색하는 커스텀 함수를 정의하고, 이를 Mistral 모델에 등록해 쿼리 처리 시 호출하게 합니다. Mistral 함수 호출에 대한 자세한 내용은 공식 [Mistral 문서]를 참조하세요.
import pandas as pd
import json
import functools
# Sample payment transaction data
data = {
'transaction_id': ['T1001', 'T1002', 'T1003', 'T1004', 'T1005'],
'customer_id': ['C001', 'C002', 'C003', 'C002', 'C001'],
'payment_amount': [125.50, 89.99, 120.00, 54.30, 210.20],
'payment_date': ['2021-10-05', '2021-10-06', '2021-10-07', '2021-10-05', '2021-10-08'],
'payment_status': ['Paid', 'Unpaid', 'Paid', 'Paid', 'Pending']
}
# Create a DataFrame from the data
df = pd.DataFrame(data)
# Function to retrieve payment status given a transaction ID
def retrieve_payment_status(df: data, transaction_id: str) -> str:
if transaction_id in df.transaction_id.values:
# Return the payment status as a JSON string
return json.dumps({'status': df[df.transaction_id == transaction_id].payment_status.item()})
return json.dumps({'error': 'transaction id not found.'})
# Function to retrieve payment date given a transaction ID
def retrieve_payment_date(df: data, transaction_id: str) -> str:
if transaction_id in df.transaction_id.values:
# Return the payment date as a JSON string
return json.dumps({'date': df[df.transaction_id == transaction_id].payment_date.item()})
return json.dumps({'error': 'transaction id not found.'})
# Define tools for the Mistral model with JSON schemas
tools = [
{
"type": "function",
"function": {
"name": "retrieve_payment_status",
"description": "Get payment status of a transaction",
"parameters": {
"type": "object",
"properties": {
"transaction_id": {
"type": "string",
"description": "The transaction id.",
}
},
"required": ["transaction_id"],
},
},
},
{
"type": "function",
"function": {
"name": "retrieve_payment_date",
"description": "Get payment date of a transaction",
"parameters": {
"type": "object",
"properties": {
"transaction_id": {
"type": "string",
"description": "The transaction id.",
}
},
"required": ["transaction_id"],
},
},
}
]
# Define tools for the Mistral model with JSON schemas
names_to_functions = {
'retrieve_payment_status': functools.partial(retrieve_payment_status, df=df),
'retrieve_payment_date': functools.partial(retrieve_payment_date, df=df)
}
tool_calling_check_transaction_status 함수는 Mistral의 함수 호출 기능을 보여줘요. 함수의 결과는 LLM의 응답에 통합되고, 이는 Langfuse에 기록·트레이싱됩니다. 이 예시는 래핑된 mistral_completion 함수를 사용해 외부 함수 호출을 Langfuse에 원활히 통합할 수 있고, 도구 선택부터 최종 출력까지 모든 단계가 완전한 관찰을 위해 포착됨을 보여줘요.
@observe()
def tool_calling_check_transaction_status(id="T1001"):
# Construct the initial user query message
messages = [{"role": "user", "content": "What's the status of my transaction {id}?".format(id=id)}]
# Use the Langfuse-decorated Mistral completion function to generate a tool-assisted response
response = mistral_completion(
model = "mistral-small-latest",
messages = messages,
max_tokens=512,
temperature=0.1,
tools = tools,
tool_choice = "any",
)
messages.append(response.choices[0].message)
# Extract the tool call details from the model's response
tool_call = response.choices[0].message.tool_calls[0]
function_name = tool_call.function.name
function_params = json.loads(tool_call.function.arguments)
# Execute the selected function with the extracted parameters
function_result = names_to_functions[function_name](**function_params)
messages.append({"role":"tool", "name":function_name, "content":function_result, "tool_call_id":tool_call.id})
# Call the Langfuse-wrapped Mistral completion function again to generate a final response using the tool's result
response = mistral_completion(
model = "mistral-small-latest",
max_tokens=1024,
temperature=0.5,
messages = messages
)
return response.choices[0].message.content
tool_calling_check_transaction_status("T1005")
피드백 (Feedback)
피드백이나 요청이 있다면 GitHub [Issue]를 만들거나 [Discord] 커뮤니티에 아이디어를 공유해 주세요.