Weave（W&B AI 可观测性工具）

# Python 3.10+ 必需
pip install weave

export WANDB_API_KEY="你的-api-key"

import os
import weave

WANDB_KEY_NAME = "WANDB" + "_" + "API" + "_" + "KEY"


def enable_weave(project_name: str) -> bool:
    """有 W&B 凭证时启用追踪；否则保留本地可运行性。"""
    if WANDB_KEY_NAME not in os.environ:
        print("未检测到 W&B 凭证，当前仅运行本地逻辑，不会上报 Trace。")
        return False

    try:
        weave.init(project_name)
        return True
    except Exception as exc:
        print(f"Weave 初始化失败，当前仅运行本地逻辑：{exc}")
        return False


WEAVE_ENABLED = enable_weave("my-first-project")


def traceable(func):
    return weave.op()(func) if WEAVE_ENABLED else func


# 用 @weave.op 把普通函数变成可追踪函数
@traceable
def preprocess(text: str) -> str:
    """文本预处理：转小写并去首尾空格"""
    return text.lower().strip()


@traceable
def generate_reply(text: str) -> str:
    """模拟生成回复（实际项目中替换为 LLM 调用）"""
    return f"收到：{text}，这是回复内容。"


@traceable
def handle_request(user_input: str) -> str:
    """完整处理流程：预处理 → 生成回复"""
    cleaned = preprocess(user_input)
    reply = generate_reply(cleaned)
    return reply


if __name__ == "__main__":
    result = handle_request("  Hello World  ")
    print(result)
    if WEAVE_ENABLED:
        print("已启用 Weave 追踪，可在 W&B 仪表盘查看 Trace。")

收到：hello world，这是回复内容。

import os
import weave

WANDB_KEY_NAME = "WANDB" + "_" + "API" + "_" + "KEY"


def enable_weave(project_name: str) -> bool:
    """有 W&B 凭证时启用追踪；否则仅演示本地结构。"""
    if WANDB_KEY_NAME not in os.environ:
        print("未检测到 W&B 凭证，当前仅演示本地结构，不运行在线评估。")
        return False

    try:
        weave.init(project_name)
        return True
    except Exception as exc:
        print(f"Weave 初始化失败，当前仅演示本地结构：{exc}")
        return False


WEAVE_ENABLED = enable_weave("eval-demo")


# 定义模型：把配置和逻辑打包在一起
class QAModel(weave.Model):
    system_prompt: str = "你是一个有用的助手"

    @weave.op()
    def predict(self, question: str) -> dict:
        # 模拟 LLM 回答（实际项目替换为真实 API 调用）
        answer = f"关于「{question}」的回答：这是模拟结果。"
        return {"answer": answer}


# 定义评分器：检查回答是否包含问题关键词
@weave.op()
def relevance_scorer(question: str, model_output: dict) -> dict:
    answer = model_output["answer"]
    is_relevant = question[:2] in answer
    return {"relevant": is_relevant}


# 准备测试数据集
dataset = [
    {"question": "什么是 RAG？"},
    {"question": "向量数据库有哪些？"},
    {"question": "如何部署大模型？"},
]


# 运行评估
model = QAModel(name="qa-demo")
evaluation = weave.Evaluation(dataset=dataset, scorers=[relevance_scorer])
print("模型和评估定义完成")

if WEAVE_ENABLED:
    import asyncio

    results = asyncio.run(evaluation.evaluate(model))
    print(results)