Skip to content

API Agent Reference

Here we handle the connection to external software tools via the parameterisation of API calls by the LLM.

The abstract base classes

BaseFetcher

Bases: ABC

Abstract base class for fetchers. A fetcher is responsible for submitting queries (in systems where submission and fetching are separate) and fetching and saving results of queries. It has to implement a fetch_results() method, which can wrap a multi-step procedure to submit and retrieve. Should implement retry method to account for connectivity issues or processing times.

Source code in biochatter/api_agent/abc.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class BaseFetcher(ABC):
    """Abstract base class for fetchers. A fetcher is responsible for submitting
    queries (in systems where submission and fetching are separate) and fetching
    and saving results of queries. It has to implement a `fetch_results()`
    method, which can wrap a multi-step procedure to submit and retrieve. Should
    implement retry method to account for connectivity issues or processing
    times.
    """

    @abstractmethod
    def fetch_results(
        self,
        query_model: BaseModel,
        retries: int | None = 3,
    ):
        """Fetches results by submitting a query. Can implement a multi-step
        procedure if submitting and fetching are distinct processes (e.g., in
        the case of long processing times as in the case of BLAST).

        Args:
        ----
            query_model: the Pydantic model describing the parameterised query

        """

fetch_results(query_model, retries=3) abstractmethod

Fetches results by submitting a query. Can implement a multi-step procedure if submitting and fetching are distinct processes (e.g., in the case of long processing times as in the case of BLAST).

query_model: the Pydantic model describing the parameterised query
Source code in biochatter/api_agent/abc.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@abstractmethod
def fetch_results(
    self,
    query_model: BaseModel,
    retries: int | None = 3,
):
    """Fetches results by submitting a query. Can implement a multi-step
    procedure if submitting and fetching are distinct processes (e.g., in
    the case of long processing times as in the case of BLAST).

    Args:
    ----
        query_model: the Pydantic model describing the parameterised query

    """

BaseInterpreter

Bases: ABC

Abstract base class for result interpreters. The interpreter is aware of the nature and structure of the results and can extract and summarise information from them.

Source code in biochatter/api_agent/abc.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class BaseInterpreter(ABC):
    """Abstract base class for result interpreters. The interpreter is aware of the
    nature and structure of the results and can extract and summarise
    information from them.
    """

    @abstractmethod
    def summarise_results(
        self,
        question: str,
        conversation_factory: Callable,
        response_text: str,
    ) -> str:
        """Summarises an answer based on the given parameters.

        Args:
        ----
            question (str): The question that was asked.

            conversation_factory (Callable): A function that creates a
                BioChatter conversation.

            response_text (str): The response.text returned from the request.

        Returns:
        -------
            A summary of the answer.

        Todo:
        ----
            Genericise (remove file path and n_lines parameters, and use a
            generic way to get the results). The child classes should manage the
            specifics of the results.

        """

summarise_results(question, conversation_factory, response_text) abstractmethod

Summarises an answer based on the given parameters.

question (str): The question that was asked.

conversation_factory (Callable): A function that creates a
    BioChatter conversation.

response_text (str): The response.text returned from the request.

A summary of the answer.
Todo:
Genericise (remove file path and n_lines parameters, and use a
generic way to get the results). The child classes should manage the
specifics of the results.
Source code in biochatter/api_agent/abc.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@abstractmethod
def summarise_results(
    self,
    question: str,
    conversation_factory: Callable,
    response_text: str,
) -> str:
    """Summarises an answer based on the given parameters.

    Args:
    ----
        question (str): The question that was asked.

        conversation_factory (Callable): A function that creates a
            BioChatter conversation.

        response_text (str): The response.text returned from the request.

    Returns:
    -------
        A summary of the answer.

    Todo:
    ----
        Genericise (remove file path and n_lines parameters, and use a
        generic way to get the results). The child classes should manage the
        specifics of the results.

    """

BaseQueryBuilder

Bases: ABC

An abstract base class for query builders.

Source code in biochatter/api_agent/abc.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class BaseQueryBuilder(ABC):
    """An abstract base class for query builders."""

    @property
    def structured_output_prompt(self) -> ChatPromptTemplate:
        """Defines a structured output prompt template. This provides a default
        implementation for an API agent that can be overridden by subclasses to
        return a ChatPromptTemplate-compatible object.
        """
        return ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "You are a world class algorithm for extracting information in structured formats.",
                ),
                (
                    "human",
                    "Use the given format to extract information from the following input: {input}",
                ),
                ("human", "Tip: Make sure to answer in the correct format"),
            ],
        )

    @abstractmethod
    def create_runnable(
        self,
        query_parameters: "BaseModel",
        conversation: "Conversation",
    ) -> Callable:
        """Creates a runnable object for executing queries. Must be implemented by
        subclasses. Should use the LangChain `create_structured_output_runnable`
        method to generate the Callable.

        Args:
        ----
            query_parameters: A Pydantic data model that specifies the fields of
                the API that should be queried.

            conversation: A BioChatter conversation object.

        Returns:
        -------
            A Callable object that can execute the query.

        """

    @abstractmethod
    def parameterise_query(
        self,
        question: str,
        conversation: "Conversation",
    ) -> BaseModel:
        """Parameterises a query object (a Pydantic model with the fields of the
        API) based on the given question using a BioChatter conversation
        instance. Must be implemented by subclasses.

        Args:
        ----
            question (str): The question to be answered.

            conversation: The BioChatter conversation object containing the LLM
                that should parameterise the query.

        Returns:
        -------
            A parameterised instance of the query object (Pydantic BaseModel)

        """

structured_output_prompt: ChatPromptTemplate property

Defines a structured output prompt template. This provides a default implementation for an API agent that can be overridden by subclasses to return a ChatPromptTemplate-compatible object.

create_runnable(query_parameters, conversation) abstractmethod

Creates a runnable object for executing queries. Must be implemented by subclasses. Should use the LangChain create_structured_output_runnable method to generate the Callable.

query_parameters: A Pydantic data model that specifies the fields of
    the API that should be queried.

conversation: A BioChatter conversation object.

A Callable object that can execute the query.
Source code in biochatter/api_agent/abc.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@abstractmethod
def create_runnable(
    self,
    query_parameters: "BaseModel",
    conversation: "Conversation",
) -> Callable:
    """Creates a runnable object for executing queries. Must be implemented by
    subclasses. Should use the LangChain `create_structured_output_runnable`
    method to generate the Callable.

    Args:
    ----
        query_parameters: A Pydantic data model that specifies the fields of
            the API that should be queried.

        conversation: A BioChatter conversation object.

    Returns:
    -------
        A Callable object that can execute the query.

    """

parameterise_query(question, conversation) abstractmethod

Parameterises a query object (a Pydantic model with the fields of the API) based on the given question using a BioChatter conversation instance. Must be implemented by subclasses.

question (str): The question to be answered.

conversation: The BioChatter conversation object containing the LLM
    that should parameterise the query.

A parameterised instance of the query object (Pydantic BaseModel)
Source code in biochatter/api_agent/abc.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@abstractmethod
def parameterise_query(
    self,
    question: str,
    conversation: "Conversation",
) -> BaseModel:
    """Parameterises a query object (a Pydantic model with the fields of the
    API) based on the given question using a BioChatter conversation
    instance. Must be implemented by subclasses.

    Args:
    ----
        question (str): The question to be answered.

        conversation: The BioChatter conversation object containing the LLM
            that should parameterise the query.

    Returns:
    -------
        A parameterised instance of the query object (Pydantic BaseModel)

    """

The API Agent

APIAgent

Source code in biochatter/api_agent/api_agent.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class APIAgent:
    def __init__(
        self,
        conversation_factory: Callable,
        query_builder: "BaseQueryBuilder",
        fetcher: "BaseFetcher",
        interpreter: "BaseInterpreter",
    ):
        """API agent class to interact with a tool's API for querying and fetching
        results.  The query fields have to be defined in a Pydantic model
        (`BaseModel`) and used (i.e., parameterised by the LLM) in the query
        builder. Specific API agents are defined in submodules of this directory
        (`api_agent`). The agent's logic is implemented in the `execute` method.

        Attributes
        ----------
            conversation_factory (Callable): A function used to create a
                BioChatter conversation, providing LLM access.

            query_builder (BaseQueryBuilder): An instance of a child of the
                BaseQueryBuilder class.

            result_fetcher (BaseFetcher): An instance of a child of the
                BaseFetcher class.

            result_interpreter (BaseInterpreter): An instance of a child of the
                BaseInterpreter class.

        """
        self.conversation_factory = conversation_factory
        self.query_builder = query_builder
        self.fetcher = fetcher
        self.interpreter = interpreter
        self.final_answer = None

    def parameterise_query(self, question: str) -> BaseModel | None:
        """Use LLM to parameterise a query (a Pydantic model) based on the given
        question using a BioChatter conversation instance.
        """
        try:
            conversation = self.conversation_factory()
            return self.query_builder.parameterise_query(question, conversation)
        except Exception as e:
            print(f"Error generating query: {e}")
            return None

    def fetch_results(self, query_model: str) -> str | None:
        """Fetch the results of the query using the individual API's implementation
        (either single-step or submit-retrieve).

        Args:
        ----
            query_model: the parameterised query Pydantic model

        """
        try:
            return self.fetcher.fetch_results(query_model, 100)
        except Exception as e:
            print(f"Error fetching results: {e}")
            return None

    def summarise_results(
        self,
        question: str,
        response_text: str,
    ) -> str | None:
        """Summarise the retrieved results to extract the answer to the question."""
        try:
            return self.interpreter.summarise_results(
                question=question,
                conversation_factory=self.conversation_factory,
                response_text=response_text,
            )
        except Exception as e:
            print(f"Error extracting answer: {e}")
            return None

    def execute(self, question: str) -> str | None:
        """Wrapper that uses class methods to execute the API agent logic. Consists
        of 1) query generation, 2) query submission, 3) results fetching, and
        4) answer extraction. The final answer is stored in the final_answer
        attribute.

        Args:
        ----
            question (str): The question to be answered.

        """
        # Generate query
        try:
            query_model = self.parameterise_query(question)
            if not query_model:
                raise ValueError("Failed to generate query.")
        except ValueError as e:
            print(e)

        # Fetch results
        try:
            response_text = self.fetch_results(
                query_model=query_model,
            )
            if not response_text:
                raise ValueError("Failed to fetch results.")
        except ValueError as e:
            print(e)

        # Extract answer from results
        try:
            final_answer = self.summarise_results(question, response_text)
            if not final_answer:
                raise ValueError("Failed to extract answer from results.")
        except ValueError as e:
            print(e)

        self.final_answer = final_answer
        return final_answer

    def get_description(self, tool_name: str, tool_desc: str):
        return f"This API agent interacts with {tool_name}'s API for querying and fetching results. {tool_desc}"

__init__(conversation_factory, query_builder, fetcher, interpreter)

API agent class to interact with a tool's API for querying and fetching results. The query fields have to be defined in a Pydantic model (BaseModel) and used (i.e., parameterised by the LLM) in the query builder. Specific API agents are defined in submodules of this directory (api_agent). The agent's logic is implemented in the execute method.

Attributes
conversation_factory (Callable): A function used to create a
    BioChatter conversation, providing LLM access.

query_builder (BaseQueryBuilder): An instance of a child of the
    BaseQueryBuilder class.

result_fetcher (BaseFetcher): An instance of a child of the
    BaseFetcher class.

result_interpreter (BaseInterpreter): An instance of a child of the
    BaseInterpreter class.
Source code in biochatter/api_agent/api_agent.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    conversation_factory: Callable,
    query_builder: "BaseQueryBuilder",
    fetcher: "BaseFetcher",
    interpreter: "BaseInterpreter",
):
    """API agent class to interact with a tool's API for querying and fetching
    results.  The query fields have to be defined in a Pydantic model
    (`BaseModel`) and used (i.e., parameterised by the LLM) in the query
    builder. Specific API agents are defined in submodules of this directory
    (`api_agent`). The agent's logic is implemented in the `execute` method.

    Attributes
    ----------
        conversation_factory (Callable): A function used to create a
            BioChatter conversation, providing LLM access.

        query_builder (BaseQueryBuilder): An instance of a child of the
            BaseQueryBuilder class.

        result_fetcher (BaseFetcher): An instance of a child of the
            BaseFetcher class.

        result_interpreter (BaseInterpreter): An instance of a child of the
            BaseInterpreter class.

    """
    self.conversation_factory = conversation_factory
    self.query_builder = query_builder
    self.fetcher = fetcher
    self.interpreter = interpreter
    self.final_answer = None

execute(question)

Wrapper that uses class methods to execute the API agent logic. Consists of 1) query generation, 2) query submission, 3) results fetching, and 4) answer extraction. The final answer is stored in the final_answer attribute.

question (str): The question to be answered.
Source code in biochatter/api_agent/api_agent.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def execute(self, question: str) -> str | None:
    """Wrapper that uses class methods to execute the API agent logic. Consists
    of 1) query generation, 2) query submission, 3) results fetching, and
    4) answer extraction. The final answer is stored in the final_answer
    attribute.

    Args:
    ----
        question (str): The question to be answered.

    """
    # Generate query
    try:
        query_model = self.parameterise_query(question)
        if not query_model:
            raise ValueError("Failed to generate query.")
    except ValueError as e:
        print(e)

    # Fetch results
    try:
        response_text = self.fetch_results(
            query_model=query_model,
        )
        if not response_text:
            raise ValueError("Failed to fetch results.")
    except ValueError as e:
        print(e)

    # Extract answer from results
    try:
        final_answer = self.summarise_results(question, response_text)
        if not final_answer:
            raise ValueError("Failed to extract answer from results.")
    except ValueError as e:
        print(e)

    self.final_answer = final_answer
    return final_answer

fetch_results(query_model)

Fetch the results of the query using the individual API's implementation (either single-step or submit-retrieve).

query_model: the parameterised query Pydantic model
Source code in biochatter/api_agent/api_agent.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def fetch_results(self, query_model: str) -> str | None:
    """Fetch the results of the query using the individual API's implementation
    (either single-step or submit-retrieve).

    Args:
    ----
        query_model: the parameterised query Pydantic model

    """
    try:
        return self.fetcher.fetch_results(query_model, 100)
    except Exception as e:
        print(f"Error fetching results: {e}")
        return None

parameterise_query(question)

Use LLM to parameterise a query (a Pydantic model) based on the given question using a BioChatter conversation instance.

Source code in biochatter/api_agent/api_agent.py 
58
59
60
61
62
63
64
65
66
67
def parameterise_query(self, question: str) -> BaseModel | None:
    """Use LLM to parameterise a query (a Pydantic model) based on the given
    question using a BioChatter conversation instance.
    """
    try:
        conversation = self.conversation_factory()
        return self.query_builder.parameterise_query(question, conversation)
    except Exception as e:
        print(f"Error generating query: {e}")
        return None

summarise_results(question, response_text)

Summarise the retrieved results to extract the answer to the question.

Source code in biochatter/api_agent/api_agent.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def summarise_results(
    self,
    question: str,
    response_text: str,
) -> str | None:
    """Summarise the retrieved results to extract the answer to the question."""
    try:
        return self.interpreter.summarise_results(
            question=question,
            conversation_factory=self.conversation_factory,
            response_text=response_text,
        )
    except Exception as e:
        print(f"Error extracting answer: {e}")
        return None

The API Agent for the BLAST tool

Module for handling BLAST API interactions.

Provides functionality for building queries, fetching results, and interpreting BLAST (Basic Local Alignment Search Tool) sequence alignment data.

BlastFetcher

Bases: BaseFetcher

A class for retrieving API results from BLAST.

Retrieves results from BLAST given a parameterised BlastQuery.

TODO add a limit of characters to be returned from the response.text?

Source code in biochatter/api_agent/blast.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
class BlastFetcher(BaseFetcher):
    """A class for retrieving API results from BLAST.

    Retrieves results from BLAST given a parameterised BlastQuery.

    TODO add a limit of characters to be returned from the response.text?
    """

    def _submit_query(self, request_data: BlastQueryParameters) -> str:
        """POST the BLAST query and retrieve the RID.

        The method submits the structured BlastQuery object and returns the RID.

        Args:
        ----
            request_data: BlastQuery object containing the BLAST query
                parameters.

        Returns:
        -------
            str: The Request ID (RID) for the submitted BLAST query.

        """
        data = {
            "CMD": request_data.cmd,
            "PROGRAM": request_data.program,
            "DATABASE": request_data.database,
            "QUERY": request_data.query,
            "FORMAT_TYPE": request_data.format_type,
            "MEGABLAST": request_data.megablast,
            "HITLIST_SIZE": request_data.max_hits,
        }
        # Include any other_params if provided
        if request_data.other_params:
            data.update(request_data.other_params)
        # Make the API call
        query_string = urlencode(data)
        # Combine base URL with the query string
        full_url = f"{request_data.url}?{query_string}"
        # Print the full URL
        request_data.full_url = full_url
        print("Full URL built by retriever:\n", request_data.full_url)
        response = requests.post(request_data.url, data=data, timeout=10)
        response.raise_for_status()
        # Extract RID from response
        print(response)
        match = re.search(r"RID = (\w+)", response.text)
        if match:
            return match.group(1)

        msg = "RID not found in BLAST submission response."
        raise ValueError(msg)

    def _fetch_results(
        self,
        rid: str,
        question_uuid: str,
        retries: int = 10000,
    ) -> str:
        """Fetch BLAST query data given RID.

        The second function to be called for a BLAST query.
        """
        ###
        ###    TO DO: Implement logging for all BLAST queries
        ###
        base_url = "https://blast.ncbi.nlm.nih.gov/Blast.cgi"
        check_status_params = {
            "CMD": "Get",
            "FORMAT_OBJECT": "SearchInfo",
            "RID": rid,
        }
        get_results_params = {
            "CMD": "Get",
            "FORMAT_TYPE": "XML",
            "RID": rid,
        }

        # Check the status of the BLAST job
        for attempt in range(retries):
            status_response = requests.get(base_url, params=check_status_params, timeout=10)
            status_response.raise_for_status()
            status_text = status_response.text
            print("evaluating status")
            if "Status=WAITING" in status_text:
                print(f"{question_uuid} results not ready, waiting...")
                time.sleep(15)
            elif "Status=FAILED" in status_text:
                msg = "BLAST query FAILED."
                raise RuntimeError(msg)
            elif "Status=UNKNOWN" in status_text:
                msg = "BLAST query expired or does not exist."
                raise RuntimeError(msg)
            elif "Status=READY" in status_text:
                if "ThereAreHits=yes" in status_text:
                    print(f"{question_uuid} results are ready, retrieving.")
                    results_response = requests.get(
                        base_url,
                        params=get_results_params,
                        timeout=10,
                    )
                    results_response.raise_for_status()
                    return results_response.text
                return "No hits found"
            if attempt == retries - 1:
                msg = "Maximum attempts reached. Results may not be ready."
                raise TimeoutError(msg)
        return None

    def fetch_results(
        self,
        query_model: BlastQueryParameters,
        retries: int = 20,
    ) -> str:
        """Submit request and fetch results from BLAST API.

        Wraps individual submission and retrieval of results.

        Args:
        ----
            query_model: the Pydantic model of the query

            retries: the number of maximum retries

        Returns:
        -------
            str: the result from the BLAST API

        """
        rid = self._submit_query(request_data=query_model)
        return self._fetch_results(
            rid=rid,
            question_uuid=query_model.question_uuid,
            retries=retries,
        )

fetch_results(query_model, retries=20)

Submit request and fetch results from BLAST API.

Wraps individual submission and retrieval of results.

query_model: the Pydantic model of the query

retries: the number of maximum retries

str: the result from the BLAST API
Source code in biochatter/api_agent/blast.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def fetch_results(
    self,
    query_model: BlastQueryParameters,
    retries: int = 20,
) -> str:
    """Submit request and fetch results from BLAST API.

    Wraps individual submission and retrieval of results.

    Args:
    ----
        query_model: the Pydantic model of the query

        retries: the number of maximum retries

    Returns:
    -------
        str: the result from the BLAST API

    """
    rid = self._submit_query(request_data=query_model)
    return self._fetch_results(
        rid=rid,
        question_uuid=query_model.question_uuid,
        retries=retries,
    )

BlastInterpreter

Bases: BaseInterpreter

A class for interpreting BLAST results.

Source code in biochatter/api_agent/blast.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
class BlastInterpreter(BaseInterpreter):
    """A class for interpreting BLAST results."""

    def summarise_results(
        self,
        question: str,
        conversation_factory: Callable,
        response_text: str,
    ) -> str:
        """Extract the answer from the BLAST results.

        Args:
        ----
            question (str): The question to be answered.
            conversation_factory: A BioChatter conversation object.
            response_text (str): The response.text returned by NCBI.

        Returns:
        -------
            str: The extracted answer from the BLAST results.

        """
        prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "You are a world class molecular biologist who knows everything about NCBI and BLAST results.",
                ),
                ("user", "{input}"),
            ],
        )
        summary_prompt = BLAST_SUMMARY_PROMPT.format(
            question=question,
            context=response_text,
        )
        output_parser = StrOutputParser()
        conversation = conversation_factory()
        chain = prompt | conversation.chat | output_parser
        return chain.invoke({"input": {summary_prompt}})

summarise_results(question, conversation_factory, response_text)

Extract the answer from the BLAST results.

question (str): The question to be answered.
conversation_factory: A BioChatter conversation object.
response_text (str): The response.text returned by NCBI.

str: The extracted answer from the BLAST results.
Source code in biochatter/api_agent/blast.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
def summarise_results(
    self,
    question: str,
    conversation_factory: Callable,
    response_text: str,
) -> str:
    """Extract the answer from the BLAST results.

    Args:
    ----
        question (str): The question to be answered.
        conversation_factory: A BioChatter conversation object.
        response_text (str): The response.text returned by NCBI.

    Returns:
    -------
        str: The extracted answer from the BLAST results.

    """
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You are a world class molecular biologist who knows everything about NCBI and BLAST results.",
            ),
            ("user", "{input}"),
        ],
    )
    summary_prompt = BLAST_SUMMARY_PROMPT.format(
        question=question,
        context=response_text,
    )
    output_parser = StrOutputParser()
    conversation = conversation_factory()
    chain = prompt | conversation.chat | output_parser
    return chain.invoke({"input": {summary_prompt}})

BlastQueryBuilder

Bases: BaseQueryBuilder

A class for building a BlastQuery object.

Source code in biochatter/api_agent/blast.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class BlastQueryBuilder(BaseQueryBuilder):
    """A class for building a BlastQuery object."""

    def create_runnable(
        self,
        query_parameters: "BlastQueryParameters",
        conversation: "Conversation",
    ) -> Callable:
        """Create a runnable object for executing queries.

        Creates a runnable using the LangChain
        `create_structured_output_runnable` method.

        Args:
        ----
            query_parameters: A Pydantic data model that specifies the fields of
                the API that should be queried.

            conversation: A BioChatter conversation object.

        Returns:
        -------
            A Callable object that can execute the query.

        """
        return create_structured_output_runnable(
            output_schema=query_parameters,
            llm=conversation.chat,
            prompt=self.structured_output_prompt,
        )

    def parameterise_query(
        self,
        question: str,
        conversation: "Conversation",
    ) -> BlastQueryParameters:
        """Generate a BlastQuery object.

        Generates the object based on the given question, prompt, and
        BioChatter conversation. Uses a Pydantic model to define the API fields.
        Creates a runnable that can be invoked on LLMs that are qualified to
        parameterise functions.

        Args:
        ----
            question (str): The question to be answered.

            conversation: The conversation object used for parameterising the
                BlastQuery.

        Returns:
        -------
            BlastQuery: the parameterised query object (Pydantic model)

        """
        runnable = self.create_runnable(
            query_parameters=BlastQueryParameters,
            conversation=conversation,
        )
        blast_call_obj = runnable.invoke(
            {"input": f"Answer:\n{question} based on:\n {BLAST_QUERY_PROMPT}"},
        )
        blast_call_obj.question_uuid = str(uuid.uuid4())
        return blast_call_obj

create_runnable(query_parameters, conversation)

Create a runnable object for executing queries.

Creates a runnable using the LangChain create_structured_output_runnable method.

query_parameters: A Pydantic data model that specifies the fields of
    the API that should be queried.

conversation: A BioChatter conversation object.

A Callable object that can execute the query.
Source code in biochatter/api_agent/blast.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def create_runnable(
    self,
    query_parameters: "BlastQueryParameters",
    conversation: "Conversation",
) -> Callable:
    """Create a runnable object for executing queries.

    Creates a runnable using the LangChain
    `create_structured_output_runnable` method.

    Args:
    ----
        query_parameters: A Pydantic data model that specifies the fields of
            the API that should be queried.

        conversation: A BioChatter conversation object.

    Returns:
    -------
        A Callable object that can execute the query.

    """
    return create_structured_output_runnable(
        output_schema=query_parameters,
        llm=conversation.chat,
        prompt=self.structured_output_prompt,
    )

parameterise_query(question, conversation)

Generate a BlastQuery object.

Generates the object based on the given question, prompt, and BioChatter conversation. Uses a Pydantic model to define the API fields. Creates a runnable that can be invoked on LLMs that are qualified to parameterise functions.

question (str): The question to be answered.

conversation: The conversation object used for parameterising the
    BlastQuery.

BlastQuery: the parameterised query object (Pydantic model)
Source code in biochatter/api_agent/blast.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def parameterise_query(
    self,
    question: str,
    conversation: "Conversation",
) -> BlastQueryParameters:
    """Generate a BlastQuery object.

    Generates the object based on the given question, prompt, and
    BioChatter conversation. Uses a Pydantic model to define the API fields.
    Creates a runnable that can be invoked on LLMs that are qualified to
    parameterise functions.

    Args:
    ----
        question (str): The question to be answered.

        conversation: The conversation object used for parameterising the
            BlastQuery.

    Returns:
    -------
        BlastQuery: the parameterised query object (Pydantic model)

    """
    runnable = self.create_runnable(
        query_parameters=BlastQueryParameters,
        conversation=conversation,
    )
    blast_call_obj = runnable.invoke(
        {"input": f"Answer:\n{question} based on:\n {BLAST_QUERY_PROMPT}"},
    )
    blast_call_obj.question_uuid = str(uuid.uuid4())
    return blast_call_obj

BlastQueryParameters

Bases: BaseModel

Pydantic model for the parameters of a BLAST query request.

The class is used for configuring and sending a request to the NCBI BLAST query API. The fields are dynamically configured by the LLM based on the user's question.

Source code in biochatter/api_agent/blast.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class BlastQueryParameters(BaseModel):
    """Pydantic model for the parameters of a BLAST query request.

    The class is used for configuring and sending a request to the NCBI BLAST
    query API. The fields are dynamically configured by the LLM based on the
    user's question.

    """

    url: str | None = Field(
        default="https://blast.ncbi.nlm.nih.gov/Blast.cgi?",
        description="ALWAYS USE DEFAULT, DO NOT CHANGE",
    )
    cmd: str | None = Field(
        default="Put",
        description="Command to execute, 'Put' for submitting query, 'Get' for retrieving results.",
    )
    program: str | None = Field(
        default="blastn",
        description=(
            "BLAST program to use, e.g., 'blastn' for nucleotide-nucleotide BLAST, "
            "'blastp' for protein-protein BLAST."
        ),
    )
    database: str | None = Field(
        default="nt",
        description=(
            "Database to search, e.g., 'nt' for nucleotide database, 'nr' for "
            "non redundant protein database, 'pdb' the Protein Data Bank "
            "database, which is used specifically for protein structures, "
            "'refseq_rna' and 'refseq_genomic': specialized databases for "
            "RNA sequences and genomic sequences"
        ),
    )
    query: str | None = Field(
        None,
        description=(
            "Nucleotide or protein sequence for the BLAST or blat query, "
            "make sure to always keep the entire sequence given."
        ),
    )
    format_type: str | None = Field(
        default="Text",
        description="Format of the BLAST results, e.g., 'Text', 'XML'.",
    )
    rid: str | None = Field(
        None,
        description="Request ID for retrieving BLAST results.",
    )
    other_params: dict | None = Field(
        default={"email": "user@example.com"},
        description="Other optional BLAST parameters, including user email.",
    )
    max_hits: int | None = Field(
        default=15,
        description="Maximum number of hits to return in the BLAST results.",
    )
    sort_by: str | None = Field(
        default="score",
        description="Criterion to sort BLAST results by, e.g., 'score', 'evalue'.",
    )
    megablast: str | None = Field(
        default="on",
        description="Set to 'on' for human genome alignemnts",
    )
    question_uuid: str | None = Field(
        default_factory=lambda: str(uuid.uuid4()),
        description="Unique identifier for the question.",
    )
    full_url: str | None = Field(
        default="TBF",
        description="Full URL to be used to submit the BLAST query",
    )

The API Agent for the OncoKB tool

OncoKBFetcher

Bases: BaseFetcher

A class for retrieving API results from OncoKB given a parameterized OncoKBQuery.

Source code in biochatter/api_agent/oncokb.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class OncoKBFetcher(BaseFetcher):
    """A class for retrieving API results from OncoKB given a parameterized
    OncoKBQuery.
    """

    def __init__(self, api_token="demo"):
        self.headers = {
            "Authorization": f"Bearer {api_token}",
            "Accept": "application/json",
        }
        self.base_url = "https://demo.oncokb.org/api/v1"

    def fetch_results(
        self,
        request_data: OncoKBQueryParameters,
        retries: int | None = 3,
    ) -> str:
        """Function to submit the OncoKB query and fetch the results directly.
        No multi-step procedure, thus no wrapping of submission and retrieval in
        this case.

        Args:
        ----
            request_data: OncoKBQuery object (Pydantic model) containing the
                OncoKB query parameters.

        Returns:
        -------
            str: The results of the OncoKB query.

        """
        # Submit the query and get the URL
        params = request_data.dict(exclude_unset=True)
        endpoint = params.pop("endpoint")
        params.pop("question_uuid")
        full_url = f"{self.base_url}/{endpoint}"
        response = requests.get(full_url, headers=self.headers, params=params)
        response.raise_for_status()

        # Fetch the results from the URL
        results_response = requests.get(response.url, headers=self.headers)
        results_response.raise_for_status()

        return results_response.text

fetch_results(request_data, retries=3)

Function to submit the OncoKB query and fetch the results directly. No multi-step procedure, thus no wrapping of submission and retrieval in this case.

request_data: OncoKBQuery object (Pydantic model) containing the
    OncoKB query parameters.

str: The results of the OncoKB query.
Source code in biochatter/api_agent/oncokb.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def fetch_results(
    self,
    request_data: OncoKBQueryParameters,
    retries: int | None = 3,
) -> str:
    """Function to submit the OncoKB query and fetch the results directly.
    No multi-step procedure, thus no wrapping of submission and retrieval in
    this case.

    Args:
    ----
        request_data: OncoKBQuery object (Pydantic model) containing the
            OncoKB query parameters.

    Returns:
    -------
        str: The results of the OncoKB query.

    """
    # Submit the query and get the URL
    params = request_data.dict(exclude_unset=True)
    endpoint = params.pop("endpoint")
    params.pop("question_uuid")
    full_url = f"{self.base_url}/{endpoint}"
    response = requests.get(full_url, headers=self.headers, params=params)
    response.raise_for_status()

    # Fetch the results from the URL
    results_response = requests.get(response.url, headers=self.headers)
    results_response.raise_for_status()

    return results_response.text

OncoKBInterpreter

Bases: BaseInterpreter

Source code in biochatter/api_agent/oncokb.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
class OncoKBInterpreter(BaseInterpreter):
    def summarise_results(
        self,
        question: str,
        conversation_factory: Callable,
        response_text: str,
    ) -> str:
        """Function to extract the answer from the BLAST results.

        Args:
        ----
            question (str): The question to be answered.
            conversation_factory: A BioChatter conversation object.
            response_text (str): The response.text returned by OncoKB.

        Returns:
        -------
            str: The extracted answer from the BLAST results.

        """
        prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "You are a world class molecular biologist who knows "
                    "everything about OncoKB and cancer genomics. Your task is "
                    "to interpret results from OncoKB API calls and summarise "
                    "them for the user.",
                ),
                ("user", "{input}"),
            ],
        )
        summary_prompt = ONCOKB_SUMMARY_PROMPT.format(
            question=question,
            context=response_text,
        )
        output_parser = StrOutputParser()
        conversation = conversation_factory()
        chain = prompt | conversation.chat | output_parser
        answer = chain.invoke({"input": {summary_prompt}})
        return answer

summarise_results(question, conversation_factory, response_text)

Function to extract the answer from the BLAST results.

question (str): The question to be answered.
conversation_factory: A BioChatter conversation object.
response_text (str): The response.text returned by OncoKB.

str: The extracted answer from the BLAST results.
Source code in biochatter/api_agent/oncokb.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
def summarise_results(
    self,
    question: str,
    conversation_factory: Callable,
    response_text: str,
) -> str:
    """Function to extract the answer from the BLAST results.

    Args:
    ----
        question (str): The question to be answered.
        conversation_factory: A BioChatter conversation object.
        response_text (str): The response.text returned by OncoKB.

    Returns:
    -------
        str: The extracted answer from the BLAST results.

    """
    prompt = ChatPromptTemplate.from_messages(
        [
            (
                "system",
                "You are a world class molecular biologist who knows "
                "everything about OncoKB and cancer genomics. Your task is "
                "to interpret results from OncoKB API calls and summarise "
                "them for the user.",
            ),
            ("user", "{input}"),
        ],
    )
    summary_prompt = ONCOKB_SUMMARY_PROMPT.format(
        question=question,
        context=response_text,
    )
    output_parser = StrOutputParser()
    conversation = conversation_factory()
    chain = prompt | conversation.chat | output_parser
    answer = chain.invoke({"input": {summary_prompt}})
    return answer

OncoKBQueryBuilder

Bases: BaseQueryBuilder

A class for building an OncoKBQuery object.

Source code in biochatter/api_agent/oncokb.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class OncoKBQueryBuilder(BaseQueryBuilder):
    """A class for building an OncoKBQuery object."""

    def create_runnable(
        self,
        query_parameters: "OncoKBQueryParameters",
        conversation: "Conversation",
    ) -> Callable:
        """Creates a runnable object for executing queries using the LangChain
        `create_structured_output_runnable` method.

        Args:
        ----
            query_parameters: A Pydantic data model that specifies the fields of
                the API that should be queried.

            conversation: A BioChatter conversation object.

        Returns:
        -------
            A Callable object that can execute the query.

        """
        return create_structured_output_runnable(
            output_schema=query_parameters,
            llm=conversation.chat,
            prompt=self.structured_output_prompt,
        )

    def parameterise_query(
        self,
        question: str,
        conversation: "Conversation",
    ) -> OncoKBQueryParameters:
        """Generates an OncoKBQuery object based on the given question, prompt, and
        BioChatter conversation. Uses a Pydantic model to define the API fields.
        Creates a runnable that can be invoked on LLMs that are qualified to
        parameterise functions.

        Args:
        ----
            question (str): The question to be answered.

            conversation: The conversation object used for parameterising the
                OncoKBQuery.

        Returns:
        -------
            OncoKBQueryParameters: the parameterised query object (Pydantic model)

        """
        runnable = self.create_runnable(
            query_parameters=OncoKBQueryParameters,
            conversation=conversation,
        )
        oncokb_call_obj = runnable.invoke(
            {"input": f"Answer:\n{question} based on:\n {ONCOKB_QUERY_PROMPT}"},
        )
        oncokb_call_obj.question_uuid = str(uuid.uuid4())
        return oncokb_call_obj

create_runnable(query_parameters, conversation)

Creates a runnable object for executing queries using the LangChain create_structured_output_runnable method.

query_parameters: A Pydantic data model that specifies the fields of
    the API that should be queried.

conversation: A BioChatter conversation object.

A Callable object that can execute the query.
Source code in biochatter/api_agent/oncokb.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def create_runnable(
    self,
    query_parameters: "OncoKBQueryParameters",
    conversation: "Conversation",
) -> Callable:
    """Creates a runnable object for executing queries using the LangChain
    `create_structured_output_runnable` method.

    Args:
    ----
        query_parameters: A Pydantic data model that specifies the fields of
            the API that should be queried.

        conversation: A BioChatter conversation object.

    Returns:
    -------
        A Callable object that can execute the query.

    """
    return create_structured_output_runnable(
        output_schema=query_parameters,
        llm=conversation.chat,
        prompt=self.structured_output_prompt,
    )

parameterise_query(question, conversation)

Generates an OncoKBQuery object based on the given question, prompt, and BioChatter conversation. Uses a Pydantic model to define the API fields. Creates a runnable that can be invoked on LLMs that are qualified to parameterise functions.

question (str): The question to be answered.

conversation: The conversation object used for parameterising the
    OncoKBQuery.

OncoKBQueryParameters: the parameterised query object (Pydantic model)
Source code in biochatter/api_agent/oncokb.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def parameterise_query(
    self,
    question: str,
    conversation: "Conversation",
) -> OncoKBQueryParameters:
    """Generates an OncoKBQuery object based on the given question, prompt, and
    BioChatter conversation. Uses a Pydantic model to define the API fields.
    Creates a runnable that can be invoked on LLMs that are qualified to
    parameterise functions.

    Args:
    ----
        question (str): The question to be answered.

        conversation: The conversation object used for parameterising the
            OncoKBQuery.

    Returns:
    -------
        OncoKBQueryParameters: the parameterised query object (Pydantic model)

    """
    runnable = self.create_runnable(
        query_parameters=OncoKBQueryParameters,
        conversation=conversation,
    )
    oncokb_call_obj = runnable.invoke(
        {"input": f"Answer:\n{question} based on:\n {ONCOKB_QUERY_PROMPT}"},
    )
    oncokb_call_obj.question_uuid = str(uuid.uuid4())
    return oncokb_call_obj