Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Python: Add Health Data example #162

Merged
merged 1 commit into from
Jan 4, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletions python/examples/healthData/demo.py
Original file line number Diff line number Diff line change
@@ -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())
63 changes: 63 additions & 0 deletions python/examples/healthData/input.txt
Original file line number Diff line number Diff line change
@@ -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

66 changes: 66 additions & 0 deletions python/examples/healthData/schema.py
Original file line number Diff line number Diff line change
@@ -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")]
74 changes: 74 additions & 0 deletions python/examples/healthData/translator.py
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading