MedAgentBench

repository·main·Indexed 43 days ago

https://github.com/stanfordmlgroup/medagentbench

A realistic virtual Electronic Health Record (EHR) environment built on AgentBench to benchmark medical Large Language Model (LLM) agents. It utilizes a FHIR server for medical data interaction and provides a pipeline for evaluating agent performance using specific GET, POST, and FINISH command formats.

Tokens
5.3K
Snippets
21
Records
24
Agent score
56%

What's inside MedAgentBench

  1. Install and set up MedAgentBench

    main

    To set up the MedAgentBench environment, follow these steps:

    1. Python Environment: Create a Conda environment with Python 3.9 and install dependencies.
    2. FHIR Server: Use Docker to pull and run the MedAgentBench image. The server must be running on port 8080.
    3. Reference Solution: Download refsol.py and place it at src/server/tasks/medagentbench/refsol.py.

    Note: This repository is intended for research purposes and may not be suitable for large-scale production.

    # Python setup
    cd MedAgentBench
    conda create -n medagentbench python=3.9
    conda activate medagentbench
    pip install -r requirements.txt
    
    # Docker setup
    docker pull jyxsu6/medagentbench:latest
    docker tag jyxsu6/medagentbench:latest medagentbench
    docker run -p 8080:8080 medagentbench
  2. Run the MedAgentBench evaluation pipeline

    main

    The evaluation process follows a specific sequence of steps:

    1. Start Task Server: Launch the task workers. The automated script src.start_task -a launches 20 workers and connects them to a controller on port 5000. Wait approximately 1 minute after running this command until you see .... 200 OK in the terminal.
    2. Start Assigner: Run the assigner to initiate the actual task tests.
    3. Retrieve Results: Once completed, results are saved to the outputs/ directory.

    Requirement: Ensure ports 5000 to 5015 are available. On macOS, you may need to free port 5000 before proceeding.

    # Step 3: Start task workers (wait 1 minute after execution)
    python -m src.start_task -a
    
    # Step 4: Start the assigner
    python -m src.assigner
  3. Configure the Agent

    main

    Before running evaluations, you must configure your LLM agent:

    • OpenAI: Add your API key to configs/agents/openai-chat.yaml.
    • Gemini/Claude (Vertex AI): Ensure you have an access token available by running gcloud auth print-access-token in your terminal.

    You can verify your configuration by running the agent_test module. By default, it uses gpt-4o-mini.

    To test a specific agent configuration, use the --config and --agent flags.

    # Verify agent configuration
    python -m src.client.agent_test --config configs/agents/api_agents.yaml --agent gpt-4o-mini
  4. MedAgentBench Prompting Format and Tool Calling Rules

    main

    The MedAgentBench task uses a specific prompt structure to guide medical LLM agents. Agents are expected to interact with a FHIR server using three specific command formats. The agent must only output one of these formats per turn and should not include any other text.

    1. GET Requests: Used to retrieve data. Format: GET url?param_name1=param_value1&param_name2=param_value2... Note: The system automatically appends &_format=json to these requests.

    2. POST Requests: Used to send data. Format:

      POST url
      [JSON payload]
    3. FINISH Command: Used to conclude the task and provide answers. Format: FINISH([answer1, answer2, ...]) Note: The content inside the parentheses must be a JSON-loadable list.

    If the agent provides an output that does not match one of these three patterns, the task will return a SampleStatus.AGENT_INVALID_ACTION status.

  5. Define ConcurrencyConfig

    main

    The ConcurrencyConfig object controls the parallel execution limits for the benchmark. It requires two dictionaries:

    • agent: A mapping of agent names (strings) to their allowed concurrency level (integers). Example: {"GPT-4": 5}.
    • task: A mapping of task names (strings) to their allowed concurrency level (integers). Example: {"diagnosis": 2}.

    Note: Any agent or task specified in the assignments must have a corresponding entry in concurrency, otherwise a validation error will be raised.

    concurrency = ConcurrencyConfig(
        agent={"agent_name": 5},
        task={"task_name": 2}
    )
  6. Define DefinitionConfig

    main

    The DefinitionConfig object maps logical agent and task names to their actual instantiation logic. It requires two dictionaries:

    • agent: A mapping of agent names to InstanceFactory objects.
    • task: A mapping of task names to InstanceFactory objects.

    These factories are used by the system to create the actual agent and task instances during the benchmark run.

    # Example mapping
    definition = DefinitionConfig(
        agent={"GPT-4": my_gpt4_factory},
        task={"diagnosis": my_diagnosis_factory}
    )
  7. Configure AssignmentConfig for MedAgentBench

    main

    The AssignmentConfig class is the primary configuration object used to define how agents and tasks are distributed and executed. It uses Pydantic for validation and supports several key components:

    Key Fields

    • assignments: A list of Assignment objects. You can provide these as a list of dictionaries where each dictionary contains agent and task keys. If a single string is provided for agent or task, it is automatically converted into a list.
    • concurrency: A ConcurrencyConfig object specifying how many instances of each agent and task should run simultaneously.
    • definition: A DefinitionConfig object that maps agent and task names to their respective InstanceFactory implementations.
    • output: A string defining the output directory. If not provided, it defaults to output/{TIMESTAMP} where {TIMESTAMP} is replaced by the current date and time (e.g., output/2023-10-27-14-30-05).

    Validation Behavior

    • Automatic Expansion: If an assignment dictionary contains a single string for agent or task, the system expands it to cover all combinations of the provided agents and tasks.
    • Unused Cleanup: The configuration automatically removes agents or tasks from definition and concurrency if they are not present in the assignments list.
    • Deduplication: If duplicate (agent, task) pairs are provided in assignments, only the first occurrence is kept, and a warning is issued.
    # Example conceptual structure of an AssignmentConfig
    config = AssignmentConfig(
        assignments=[
            {"agent": "GPT-4", "task": "diagnosis"},
            {"agent": ["Llama-3", "Claude-3"], "task": "treatment"}
        ],
        concurrency={
            "agent": {"GPT-4": 2, "Llama-3": 4, "Claude-3": 4},
            "task": {"diagnosis": 1, "treatment": 1}
        },
        definition={
            "agent": {"GPT-4": factory_gpt4, "Llama-3": factory_llama, "Claude-3": factory_claude},
            "task": {"diagnosis": factory_diag, "treatment": factory_treat}
        },
        output="my_results/{TIMESTAMP}"
    )
  8. Verify FHIR server connectivity with verify_fhir_server

    main

    Use verify_fhir_server(fhir_api_base) to check if a connection to a FHIR server is functional. The function attempts to fetch the server metadata by sending a GET request to {fhir_api_base}/metadata. It returns True if the server responds with a 200 status code, and False otherwise.

    from src.server.tasks.medagentbench.utils import verify_fhir_server
    
    is_working = verify_fhir_server('https://your-fhir-server-url.com/api/v1/')
    if is_working:
        print("FHIR server is reachable.")
    else:
        print("FHIR server connection failed.")
  9. Calculate MedAgentBench overall success rate

    main

    After running multiple samples, use the calculate_overall method to evaluate the performance of the agent. This method compares the agent's result against the ground truth using the eval function (which relies on the refsol.py module).

    Input:

    • results: A list of TaskOutput objects generated from the start_sample method.

    Output: Returns a dictionary containing:

    • success rate: The ratio of correct tasks to total tasks.
    • raw_results: The original list of TaskOutput objects, with their status field updated to include 'Correct' or 'Incorrect'.
    # Assuming 'results' is a list of TaskOutput from start_sample
    summary = med_agent_bench_task.calculate_overall(results)
    print(f"Success Rate: {summary['success rate']}")
  10. Implement the MedAgentBench Task class

    main

    To use MedAgentBench as a task within the AgentBench framework, you must instantiate the MedAgentBench class. This class requires a configuration dictionary containing specific keys to load the dataset, available functions, and the FHIR API endpoint.

    Required Configuration Keys:

    • data_file: Path to the JSON file containing the task data (context and instructions).
    • func_file: Path to the JSON file containing the list of available FHIR functions.
    • fhir_api_base: The base URL for the FHIR server. The class will attempt to verify this connection upon initialization.

    Optional Configuration Keys:

    • max_round: The maximum number of tool-calling rounds allowed per task (defaults to 5).

    Important Setup Note: This task requires a reference solution file located at src/server/tasks/medagentbench/refsol.py to perform evaluations. If this file is missing, the task will fail to initialize.

    # Example initialization (conceptual)
    configs = {
        "data_file": "path/to/data.json",
        "func_file": "path/to/functions.json",
        "fhir_api_base": "http://your-fhir-server:8080/fhir",
        "max_round": 10
    }
    med_agent_bench_task = MedAgentBench(**configs)
  11. Send GET requests with send_get_request

    main

    The send_get_request utility performs an HTTP GET request and returns a standardized dictionary containing the response.

    Arguments:

    • url (str): The target URL.
    • params (dict, optional): Query parameters.
    • headers (dict, optional): HTTP headers.

    Returns:

    • A dictionary with keys status_code and data (if successful).
    • A dictionary with key error (if an exception occurs).

    Note: If the response Content-Type is application/json, the data field will contain the parsed JSON object; otherwise, it contains the raw response text.

    from src.server.tasks.medagentbench.utils import send_get_request
    
    # Example: Successful JSON request
    result = send_get_request('https://api.example.com/resource', params={'id': '123'})
    if 'error' not in result:
        print(f"Status: {result['status_code']}", result['data'])
    else:
        print(f"Error occurred: {result['error']}")