用 Llama 4 评测 LlamaStack Web Search Groundedness

在本教程中，我们将衡量 LlamaStack web search agent 生成响应的 groundedness。LlamaStack 是 Meta 维护的开源框架，用于简化基于大型语言模型的应用开发与部署。评测将使用 Ragas 指标，并以 Meta Llama 4 Maverick 作为 judge。

设置并运行 LlamaStack 服务器

该命令安装 LlamaStack 服务器与 together inference provider 所需的全部依赖

使用 conda 时的命令

!pip install ragas langchain-together uv 
!uv run --with llama-stack llama stack build --template together --image-type conda

使用 venv 时的命令

!pip install ragas langchain-together uv 
!uv run --with llama-stack llama stack build --template together --image-type venv

import os
import subprocess


def run_llama_stack_server_background():
    log_file = open("llama_stack_server.log", "w")
    process = subprocess.Popen(
        "uv run --with llama-stack llama stack run together --image-type venv",
        shell=True,
        stdout=log_file,
        stderr=log_file,
        text=True,
    )

    print(f"Starting LlamaStack server with PID: {process.pid}")
    return process


def wait_for_server_to_start():
    import requests
    from requests.exceptions import ConnectionError
    import time

    url = "http://0.0.0.0:8321/v1/health"
    max_retries = 30
    retry_interval = 1

    print("Waiting for server to start", end="")
    for _ in range(max_retries):
        try:
            response = requests.get(url)
            if response.status_code == 200:
                print("\nServer is ready!")
                return True
        except ConnectionError:
            print(".", end="", flush=True)
            time.sleep(retry_interval)

    print("\nServer failed to start after", max_retries * retry_interval, "seconds")
    return False


# use this helper if needed to kill the server
def kill_llama_stack_server():
    # Kill any existing llama stack server processes
    os.system(
        "ps aux | grep -v grep | grep llama_stack.distribution.server.server | awk '{print $2}' | xargs kill -9"
    )

启动 LlamaStack 服务器

server_process = run_llama_stack_server_background()
assert wait_for_server_to_start()

Starting LlamaStack server with PID: 95508
Waiting for server to start....
Server is ready!

构建 Search Agent

from llama_stack_client import LlamaStackClient, Agent, AgentEventLogger

client = LlamaStackClient(
    base_url="http://0.0.0.0:8321",
)

agent = Agent(
    client,
    model="meta-llama/Llama-3.1-8B-Instruct",
    instructions="You are a helpful assistant. Use web search tool to answer the questions.",
    tools=["builtin::websearch"],
)
user_prompts = [
    "In which major did Demis Hassabis complete his undergraduate degree? Search the web for the answer.",
    "Ilya Sutskever is one of the key figures in AI. From which institution did he earn his PhD in machine learning? Search the web for the answer.",
    "Sam Altman, widely known for his role at OpenAI, was born in which American city? Search the web for the answer.",
]

session_id = agent.create_session("test-session")


for prompt in user_prompts:
    response = agent.create_turn(
        messages=[
            {
                "role": "user",
                "content": prompt,
            }
        ],
        session_id=session_id,
    )
    for log in AgentEventLogger().log(response):
        log.print()

现在，让我们更深入地查看 agent 的执行步骤，看看 agent 表现如何。

session_response = client.agents.session.retrieve(
    session_id=session_id,
    agent_id=agent.agent_id,
)

评测 Agent 响应

我们想衡量 LlamaStack web search Agent 生成响应的 Groundedness。为此需要 EvaluationDataset 以及用于评估 grounded response 的指标。Ragas 提供大量开箱即用的指标，可用于衡量检索和生成的各个方面。

衡量响应 groundedness 时我们将使用：

1. Faithfulness
2. Response Groundedness

构建 Ragas EvaluationDataset

要用 Ragas 进行评测，我们将创建一个 EvaluationDataset

import json

# This function extracts the search results for the trace of each query
def extract_retrieved_contexts(turn_object):
    results = []
    for step in turn_object.steps:
        if step.step_type == "tool_execution":
            tool_responses = step.tool_responses
            for response in tool_responses:
                content = response.content
                if content:
                    try:
                        parsed_result = json.loads(content)
                        results.append(parsed_result)
                    except json.JSONDecodeError:
                        print("Warning: Unable to parse tool response content as JSON.")
                        continue

    retrieved_context = []
    for result in results:
        top_content_list = [item["content"] for item in result["top_k"]]
        retrieved_context.extend(top_content_list)
    return retrieved_context

from ragas.dataset_schema import EvaluationDataset

samples = []

references = [
    "Demis Hassabis completed his undergraduate degree in Computer Science.",
    "Ilya Sutskever earned his PhD from the University of Toronto.",
    "Sam Altman was born in Chicago, Illinois.",
]

for i, turn in enumerate(session_response.turns):
    samples.append(
        {
            "user_input": turn.input_messages[0].content,
            "response": turn.output_message.content,
            "reference": references[i],
            "retrieved_contexts": extract_retrieved_contexts(turn),
        }
    )

ragas_eval_dataset = EvaluationDataset.from_list(samples)

ragas_eval_dataset.to_pandas()

设置 Ragas 指标

from ragas.metrics import AnswerAccuracy, Faithfulness, ResponseGroundedness
from langchain_together import ChatTogether
from ragas.llms import LangchainLLMWrapper

llm = ChatTogether(
    model="meta-llama/Llama-4-Maverick-17B-128E-Instruct-FP8",
)
evaluator_llm = LangchainLLMWrapper(llm)

ragas_metrics = [
    AnswerAccuracy(llm=evaluator_llm),
    Faithfulness(llm=evaluator_llm),
    ResponseGroundedness(llm=evaluator_llm),
]

评测

最后，让我们运行评测。

from ragas import evaluate

results = evaluate(dataset=ragas_eval_dataset, metrics=ragas_metrics)
results.to_pandas()

Evaluating: 100%|██████████| 9/9 [00:04<00:00,  2.03it/s]

kill_llama_stack_server()