# Copyright (c) Microsoft. All rights reserved. from __future__ import annotations import logging from typing import Any, Dict, List, cast import pandas as pd from agents import Agent, Runner from agents.extensions.models.litellm_model import LitellmModel from agents.mcp import MCPServerSse from agents.model_settings import ModelSettings from metric_utils import compute_scores import agentlightning as agl logger = logging.getLogger("rag_agent") agent_prompt = """You are an assistant who answers questions using Wikipedia retriever. Answer the question using only the retrieved passages. Verify your answer directly against the text. After each search: - Summarize findings. - Decide if info is sufficient. - If sufficient: reply in ... with your answer. The answer must be extremely concise: a single word or a few words only. - If not: suggest the next search needed to fill info gaps. The system will return top 3 relevant Wikipedia chunks. - Explain your reasoning for the chosen action. Repeat as needed. When done, wrap your final, concise answer in tags.""" class RAGAgent(agl.LitAgent[Dict[str, Any]]): """RAGAgent is an agent that relies on a MCP-based retriever to answer questions.""" def __init__(self) -> None: super().__init__() self.mcp_server_url = "http://127.0.0.1:8099/sse" async def training_rollout_async( self, task: Dict[str, Any], resources: agl.NamedResources, rollout: agl.Rollout ) -> float | None: # llm resources llm = cast(agl.LLM, resources["main_llm"]) # The rollout should carry an attempt inside rollout = cast(agl.AttemptedRollout, rollout) base_url = llm.get_base_url(rollout.rollout_id, rollout.attempt.attempt_id) logger.info(f"Training with model: {llm.model} on endpoint: {base_url}") async with MCPServerSse( name="wiki_retriever_mcp", params={"url": self.mcp_server_url}, ) as server: agent = Agent( model=LitellmModel( model="hosted_vllm/" + llm.model, base_url=base_url, ), model_settings=ModelSettings( max_tokens=2048, temperature=0.7, ), name="Assistant", instructions=agent_prompt, mcp_servers=[server], ) result = await Runner.run(agent, task["question"]) answer = result.final_output # reward reward = compute_scores(answer, str(task["answer"])) logger.info( "Question: %s\nAnswer: %s\nGround truth: %s\nReward: %s", task["question"], answer, task["answer"], reward, ) return float(reward) # Convert to float for compatibility with the Runner async def validation_rollout_async( self, task: Dict[str, Any], resources: agl.NamedResources, rollout: agl.Rollout ) -> float | None: """Validation rollout will share the same logic as the training rollout.""" # Same as training rollout, but with different temperature llm = cast(agl.LLM, resources["main_llm"]) rollout = cast(agl.AttemptedRollout, rollout) # set temperature val_resources: agl.NamedResources = { "main_llm": agl.LLM( endpoint=llm.get_base_url(rollout.rollout_id, rollout.attempt.attempt_id), model=llm.model, sampling_parameters={"temperature": 0.7}, ) } # reuse training rollout for validation return await self.training_rollout_async(task, val_resources, rollout) def debug(): """Debug the RAGAgent.""" agl.setup_logging("DEBUG", apply_to=[logger.name]) # 1. loading dataset dataset_path = "data/dataset_tiny.parquet" df: pd.DataFrame = pd.read_parquet(dataset_path) # type: ignore data: List[Dict[str, Any]] = df.head(5).to_dict(orient="records") # type: ignore # NOTE: The following dummy data can also be used if you don't have the dataset. # data: List[Dict[str, Any]] = [{"question": "What is the capital of France?", "answer": "Paris"}] # 2. configuring resources (LLM) # Note: You need to start a local service compatible with the OpenAI API (such as vLLM) # For example: python -m vllm.entrypoints.openai.api_server --model Qwen/Qwen2.5-1.5B-Instruct --port 8000 resources: dict[str, agl.ResourceUnion] = { "main_llm": agl.LLM( endpoint="http://localhost:8000/v1", # Replace with your actual vLLM address model="Qwen/Qwen2.5-1.5B-Instruct", # Replace with your actual loaded model name sampling_parameters={"temperature": 0.0}, ) } # 3. run agent trainer = agl.Trainer(initial_resources=resources) trainer.dev(RAGAgent(), train_dataset=data) # type: ignore if __name__ == "__main__": debug()