Exa 集成

本指南演示如何将 Litefuse 与 Exa 集成，以追踪你的 AI 搜索操作。借助 Litefuse 的追踪能力，你可以自动捕获 Exa 搜索函数的输入、输出和执行时间等细节。

什么是 Exa？ Exa 是一个为 LLM 与 AI 应用打造的 AI 搜索 API。与传统搜索引擎不同，Exa 旨在理解语义并检索高质量、相关的结果，非常适合 RAG（检索增强生成）、研究、内容发现等 AI 场景。

什么是 Litefuse？ Litefuse 是一个开源的 AI Agent 可观测性与评估平台，帮助团队追踪 API 调用、监控性能并调试 AI 应用中的问题。

安装依赖

首先，安装必要的 Python 包：

%pip install langfuse exa-py

设置环境变量

通过注册 Litefuse Cloud 或自托管 Litefuse 来获取 Litefuse API Key。你还需要准备 Exa 与 OpenAI 的 API Key。

import os
 
# 从项目设置页面获取你项目的 Key：https://litefuse.cloud
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-..." 
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-..." 
os.environ["LANGFUSE_HOST"] = "https://litefuse.cloud"
 
# 你的 Exa API key
os.environ["EXA_API_KEY"] = "..."
 
# 你的 openai key
os.environ["OPENAI_API_KEY"] = "sk-..."

设置好环境变量后，我们就可以初始化 Exa 与 Langfuse 客户端了。

from exa_py import Exa
from langfuse import get_client
 
exa = Exa(api_key= os.environ["EXA_API_KEY"])
langfuse = get_client()

示例 1：追踪 Exa `search_and_contents`

我们使用 Litefuse @observe() 装饰器监控 Exa 搜索操作。本示例中，@observe() 装饰器会捕获 search_with_exa() 函数的输入、输出和执行时间。若想更精细地控制发送到 Litefuse 的数据，可以使用 Python SDK 的 Context Manager 或手动创建 observation。

from langfuse import observe
 
@observe(as_type="retriever")
def search_with_exa(query: str, num_results: int = 5):
    """Search the web using Exa AI and return results."""
    results = exa.search_and_contents(
        query,
        num_results=num_results,
        text=True
    )
    return results
 
# 示例：搜索关于 Litefuse 的信息
search_results = search_with_exa("What is Litefuse and how does it help with LLM observability?")
 
# 输出结果
for result in search_results.results:
    print(f"Title: {result.title}")
    print(f"URL: {result.url}")
    print(f"Text: {result.text[:200]}...\n")

示例 2：Exa 搜索结合 OpenAI

你也可以追踪更复杂的工作流，比如用 OpenAI 总结搜索结果。这里我们使用 Litefuse @observe() 装饰器把 Exa 搜索和 OpenAI 生成合并到同一个 trace 中。

import os
from langfuse.openai import OpenAI
 
@observe()
def search_and_summarize(query: str):
 
    # 1. Exa 搜索
    @observe(as_type="retriever")
    def search_with_exa(query: str, num_results: int = 5):
        """Search the web using Exa AI and return results."""
        results = exa.search_and_contents(
            query,
            num_results=num_results,
            text=True
        )
        return results
 
    results = search_with_exa(query)
 
    # 2. 构造简短的上下文
    context = "\n".join([f"{r.title} ({r.url}): {r.text}" for r in results.results])
 
    # 3. 用 OpenAI 进行总结
    client = OpenAI()
    resp = client.chat.completions.create(
        model="gpt-5-mini",
        messages=[
            {"role": "system", "content": "Summarize the following search results clearly and concisely."},
            {"role": "user", "content": context}
        ]
    )
 
    print("Summary:\n", resp.choices[0].message.content)
 
search_and_summarize("What is Litefuse and how does it help with AI agent observability?")

在 Litefuse 中查看 trace

执行完上述被追踪的函数后，登录你的 Litefuse 仪表盘查看详细的 trace 日志。你将能看到：

搜索查询及其参数
每次 API 调用的响应耗时
显示搜索与相似性操作关系的嵌套 trace
用于调试的完整输入输出数据

Litefuse 中的示例 trace

Interoperability with the Python SDK

You can use this integration together with the Litefuse SDKs to add additional attributes to the observation.

The @observe() decorator provides a convenient way to automatically wrap your instrumented code and add additional attributes to the observation.

from langfuse import observe, propagate_attributes, get_client
 
langfuse = get_client()
 
@observe()
def my_llm_pipeline(input):
    # Add additional attributes (user_id, session_id, metadata, version, tags) to all spans created within this execution scope
    with propagate_attributes(
        user_id="user_123",
        session_id="session_abc",
        tags=["agent", "my-observation"],
        metadata={"email": "user@litefuse.ai"},
        version="1.0.0"
    ):
 
        # YOUR APPLICATION CODE HERE
        result = call_llm(input)
 
        return result
 
# Run the function
my_llm_pipeline("Hi")

Learn more about using the Decorator in the Langfuse SDK instrumentation docs.

Troubleshooting

No observations appearing

First, enable debug mode in the Python SDK:

export LANGFUSE_DEBUG="True"

Then run your application and check the debug logs:

OTel observations appear in the logs: Your application is instrumented correctly but observations are not reaching Litefuse. To resolve this:
1. Call langfuse.flush() at the end of your application to ensure all observations are exported.
2. Verify that you are using the correct API keys and base URL.
No OTel spans in the logs: Your application is not instrumented correctly. Make sure the instrumentation runs before your application code.

Unwanted observations in Litefuse

The Langfuse SDK is based on OpenTelemetry. Other libraries in your application may emit OTel spans that are not relevant to you. These still count toward your billable units, so you should filter them out. See Unwanted spans in Litefuse for details.

Missing attributes

Some attributes may be stored in the metadata object of the observation rather than being mapped to the Litefuse data model. If a mapping or integration does not work as expected, please raise an issue on GitHub.