From 10a7772aa15f0050a50807470469c5eb7764db46 Mon Sep 17 00:00:00 2001 From: Hillary Mutisya Date: Thu, 4 Jan 2024 11:30:46 -0800 Subject: [PATCH] Add Health Data example --- python/examples/healthData/demo.py | 53 +++++++++ python/examples/healthData/input.txt | 63 ++++++++++ python/examples/healthData/schema.py | 66 +++++++++++ python/examples/healthData/translator.py | 74 ++++++++++++ python/notebooks/healthData.ipynb | 145 +++++++++++++++++++++++ 5 files changed, 401 insertions(+) create mode 100644 python/examples/healthData/demo.py create mode 100644 python/examples/healthData/input.txt create mode 100644 python/examples/healthData/schema.py create mode 100644 python/examples/healthData/translator.py create mode 100644 python/notebooks/healthData.ipynb diff --git a/python/examples/healthData/demo.py b/python/examples/healthData/demo.py new file mode 100644 index 00000000..e52655f6 --- /dev/null +++ b/python/examples/healthData/demo.py @@ -0,0 +1,53 @@ +import asyncio +import json +import sys +from dotenv import dotenv_values +import schema as health +from typechat import Failure, TypeChatValidator, create_language_model +from translator import TranslatorWithHistory + +health_instructions = """ +Help me enter my health data step by step. +Ask specific questions to gather required and optional fields I have not already providedStop asking if I don't know the answer +Automatically fix my spelling mistakes +My health data may be complex: always record and return ALL of it. +Always return a response: +- If you don't understand what I say, ask a question. +- At least respond with an OK message. + +""" + + +async def main(): + vals = dotenv_values() + model = create_language_model(vals) + validator = TypeChatValidator(health.HealthDataResponse) + translator = TranslatorWithHistory( + model, validator, health.HealthDataResponse, additional_agent_instructions=health_instructions + ) + print("šŸ’‰šŸ’ŠšŸ¤§> ", end="", flush=True) + for line in sys.stdin: + result = await translator.translate(line) + if isinstance(result, Failure): + print("Translation Failed āŒ") + print(f"Context: {result.message}") + else: + result = result.value + print("Translation Succeeded! āœ…\n") + print("JSON View") + print(json.dumps(result, indent=2)) + + message = result.get("message", None) + not_translated = result.get("notTranslated", None) + + if message: + print(f"\nšŸ“: {message}" ) + + if not_translated: + print(f"\nšŸ¤”: I did not understand\n {not_translated}" ) + + print("\nšŸ’‰šŸ’ŠšŸ¤§> ", end="", flush=True) + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/python/examples/healthData/input.txt b/python/examples/healthData/input.txt new file mode 100644 index 00000000..21138470 --- /dev/null +++ b/python/examples/healthData/input.txt @@ -0,0 +1,63 @@ +# +# Conversations with a Health Data Agent +# For each conversation: +# You start with the first line +# Then type the next line in response +# + +# ================ +# USE GPT4 +# ================ +# Conversation: +i want to record my shingles +August 2016 +It lasted 3 months +I also broke my foot +I broke it in high school +2001 +The foot took a year to be ok + +# Conversation: +klaritin +2 tablets 3 times a day +300 mg +actually that is 1 tablet +@clear + +# Conversation: +klaritin +1 pill, morning and before bedtime +Can't remember +Actually, that is 3 tablets +500 mg +@clear + +#Conversation +I am taking binadryl now +As needed. Groceery store strength +That is all I have +I also got allergies. Pollen +@clear + +# Conversation: +Robotussin +1 cup +Daily, as needed +Robotussin with Codeine +Put down strength as I don't know +@clear + +# Conversation: +Hey +Melatonin +1 3mg tablet every night +@clear + +# Conversation: +I got the flu +Started 2 weeks ago +Its gone now. Only lasted about a week +I took some sudafed though +I took 2 sudafed twice a day. Regular strength +@clear + diff --git a/python/examples/healthData/schema.py b/python/examples/healthData/schema.py new file mode 100644 index 00000000..360e05a3 --- /dev/null +++ b/python/examples/healthData/schema.py @@ -0,0 +1,66 @@ +from typing import TypedDict, Annotated, NotRequired, Literal + + +def Doc(s: str) -> str: + return s + + +class Quantity(TypedDict): + value: Annotated[float, Doc("Exact number")] + units: Annotated[str, Doc("UNITS include mg, kg, cm, pounds, liter, ml, tablet, pill, cup, per-day, per-week..ETC")] + + +class ApproxDatetime(TypedDict): + displayText: Annotated[str, Doc("Default: Unknown. Required")] + timestamp: NotRequired[Annotated[str, Doc("If precise timestamp can be set")]] + + +class ApproxQuantity(TypedDict): + displayText: Annotated[str, Doc("Default: Unknown. Required")] + quantity: NotRequired[Annotated[Quantity, Doc("Optional: only if precise quantities are available")]] + + +class OtherHealthData(TypedDict): + """ + Use for health data that match nothing else. E.g. immunization, blood prssure etc + """ + + text: str + when: NotRequired[ApproxDatetime] + + +class Condition(TypedDict): + """ + Disease, Ailment, Injury, Sickness + """ + + name: Annotated[str, Doc("Fix any spelling mistakes, especially phonetic spelling")] + startDate: Annotated[ApproxDatetime, Doc("When the condition started? Required")] + status: Annotated[ + Literal["active", "recurrence", "relapse", "inactive", "remission", "resolved", "unknown"], + Doc("Always ask for current status of the condition"), + ] + endDate: NotRequired[Annotated[ApproxDatetime, Doc("If the condition was no longer active")]] + + +class Medication(TypedDict): + """ + Meds, pills etc. + """ + + name: Annotated[str, Doc("Fix any spelling mistakes, especially phonetic spelling")] + dose: Annotated[ApproxQuantity, Doc("E.g. 2 tablets, 1 cup. Required")] + frequency: Annotated[ApproxQuantity, Doc("E.g. twice a day. Required")] + strength: Annotated[ApproxQuantity, Doc("E.g. 50 mg. Required")] + + +class HealthData(TypedDict, total=False): + medication: list[Medication] + condition: list[Condition] + other: list[OtherHealthData] + + +class HealthDataResponse(TypedDict, total=False): + data: Annotated[HealthData, Doc("Return this if JSON has ALL required information. Else ask questions")] + message: Annotated[str, Doc("Use this to ask questions and give pertinent responses")] + notTranslated: Annotated[str, Doc("Use this parts of the user request not translateed, off topic, etc")] diff --git a/python/examples/healthData/translator.py b/python/examples/healthData/translator.py new file mode 100644 index 00000000..7aaaa3b5 --- /dev/null +++ b/python/examples/healthData/translator.py @@ -0,0 +1,74 @@ +import json +from textwrap import dedent, indent +from typing import TypeVar, Any, override, TypedDict, Literal + +from typechat import TypeChatValidator, TypeChatModel, TypeChatTranslator, Result, Failure + +from datetime import datetime + +T = TypeVar("T", covariant=True) + + +class ChatMessage(TypedDict): + source: Literal["system", "user", "assistant"] + body: Any + + +class TranslatorWithHistory(TypeChatTranslator[T]): + _chat_history: list[ChatMessage] + _max_prompt_length: int + _additional_agent_instructions: str + + def __init__( + self, model: TypeChatModel, validator: TypeChatValidator[T], target_type: type[T], additional_agent_instructions: str + ): + super().__init__(model=model, validator=validator, target_type=target_type) + self._chat_history = [] + self._max_prompt_length = 2048 + self._additional_agent_instructions = additional_agent_instructions + + @override + async def translate(self, request: str) -> Result[T]: + result = await super().translate(request=request) + if not isinstance(result, Failure): + self._chat_history.append(ChatMessage(source="assistant", body=result.value)) + return result + + @override + def _create_request_prompt(self, intent: str) -> str: + # TODO: drop history entries if we exceed the max_prompt_length + history_str = json.dumps(self._chat_history, indent=2, default=lambda o: None, allow_nan=False) + history_str = indent(history_str, " ") + + schema_str = indent(self._schema_str, " ") + + instructions_str = indent(self._additional_agent_instructions, " ") + + now = datetime.now() + + prompt = F""" + user: You are a service that translates user requests into JSON objects of type "{self._type_name}" according to the following TypeScript definitions: + ''' + {schema_str} + ''' + + user: + Use precise date and times RELATIVE TO CURRENT DATE: {now.strftime('%A, %m %d, %Y')} CURRENT TIME: {now.strftime("%H:%M:%S")} + Also turn ranges like next week and next month into precise dates + + user: + {instructions_str} + + system: + IMPORTANT CONTEXT for the user request: + {history_str} + + user: + The following is a user request: + ''' + {intent} + ''' + The following is the user request translated into a JSON object with 2 spaces of indentation and no properties with the value undefined: + """ + prompt = dedent(prompt) + return prompt diff --git a/python/notebooks/healthData.ipynb b/python/notebooks/healthData.ipynb new file mode 100644 index 00000000..2bfa7b45 --- /dev/null +++ b/python/notebooks/healthData.ipynb @@ -0,0 +1,145 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "!pip install --upgrade setuptools\n", + "!pip install --upgrade gradio" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import json\n", + "import setuptools\n", + "\n", + "import os\n", + "import sys\n", + "module_path = os.path.abspath(os.path.join('..'))\n", + "if module_path not in sys.path:\n", + " sys.path.append(module_path)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from dotenv import dotenv_values\n", + "from typechat import Failure, TypeChatValidator, create_language_model\n", + "from examples.healthData import schema as health\n", + "from examples.healthData.translator import TranslatorWithHistory" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "health_instructions = \"\"\"\n", + "Help me enter my health data step by step.\n", + "Ask specific questions to gather required and optional fields I have not already providedStop asking if I don't know the answer\n", + "Automatically fix my spelling mistakes\n", + "My health data may be complex: always record and return ALL of it.\n", + "Always return a response:\n", + "- If you don't understand what I say, ask a question.\n", + "- At least respond with an OK message.\n", + "\n", + "\"\"\"\n", + "\n", + "vals = dotenv_values()\n", + "model = create_language_model(vals)\n", + "validator = TypeChatValidator(health.HealthDataResponse)\n", + "translator = TranslatorWithHistory(model, validator, health.HealthDataResponse, additional_agent_instructions=health_instructions)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import pandas\n", + "\n", + "async def get_translation(message, history):\n", + " result = await translator.translate(message)\n", + " if isinstance(result, Failure):\n", + " return f\"Translation Failed āŒ \\n Context: {result.message}\"\n", + " else:\n", + " result = result.value\n", + " output = f\"Translation Succeeded! āœ…\\n\"\n", + " \n", + " data = result.get(\"data\", None)\n", + " if data:\n", + " df = pandas.DataFrame.from_dict(data)\n", + " output += f\"HealthData \\n ``` {df.fillna('').to_markdown(tablefmt='grid')} \\n ``` \\n\"\n", + "\n", + " message = result.get(\"message\", None)\n", + " not_translated = result.get(\"notTranslated\", None)\n", + "\n", + " if message:\n", + " output += f\"\\nšŸ“: {message}\"\n", + " \n", + " if not_translated:\n", + " output += f\"\\nšŸ¤”: I did not understand\\n {not_translated}\" \n", + " \n", + " return output\n", + "\n", + "\n", + "def get_examples():\n", + " example_prompts = []\n", + " with open('../examples/healthData/input.txt') as prompts_file:\n", + " for line in prompts_file:\n", + " example_prompts.append(line)\n", + " return example_prompts\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import gradio as gr\n", + "\n", + "gr.ChatInterface(get_translation, title=\"šŸ’‰šŸ’ŠšŸ¤§ Health Data\", examples=get_examples()).launch(debug=False)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3 (ipykernel)", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.12.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +}