Spaces:
Sleeping
Sleeping
File size: 7,742 Bytes
90f1f18 3e595a5 90f1f18 3e595a5 90f1f18 3e595a5 90f1f18 3e595a5 90f1f18 3e595a5 90f1f18 3e595a5 90f1f18 3e595a5 |
1 2 3 4 5 6 7 8 9 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 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 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 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 |
import pprint
from langchain_core.tools import tool
from modules.data_class import DataState
from langgraph.prebuilt import InjectedState
from langchain_core.messages.tool import ToolMessage
# These functions have no body; LangGraph does not allow @tools to update
# the conversation state, so you will implement a separate node to handle
# state updates. Using @tools is still very convenient for defining the tool
# schema, so empty functions have been defined that will be bound to the LLM
# but their implementation is deferred to the order_node.
@tool
def patient_id(name: str, DOB: str, gender: str, contact: str, emergency_contact: str) -> str:
"""Collecting basic patient identification information including:
- Basic information (name, DOB, gender, contact details)
- Emergency contact information
Returns:
The updated data with the patient ID information added.
"""
@tool
def symptom(main_symptom: str, symptom_length: str) -> str:
"""Collecting patient's main symptom assessment including:
- Primary symptoms
- Duration of the symptoms
Returns:
The updated data with the patient's symptom information added.
"""
@tool
def pain(pain_location: str, pain_side: str, pain_intensity: int, pain_description: str, start_time: str, radiation: bool, triggers: str, symptom: str) -> str:
"""Collecting patient's pain status including:
- Pain location using body mapping (head, arms, hands, trunk, legs, feet)
- Pain side (left or right)
- Pain intensity (0-10 scale for each location)
- Pain characteristics and patterns
- Onset time
- Radiation patterns
- Triggering factors
- Associated symptoms
Returns:
The updated data with the patient's pain status added.
"""
@tool
def medical_hist(medical_condition: str, first_time: str, surgery_history: list, medication: str, allergy: str) -> str:
"""Collecting patient's medical history including:
- Existing medical conditions
- First occurrence date
- Surgical history with dates
- Current medications
- Allergies
Returns:
The updated data with the patient's medical history added.
"""
@tool
def family_hist(family_history: str) -> str:
"""Collecting patient's family history
Returns:
The updated data with the patient's family history added.
"""
@tool
def social_hist(occupation: str, smoking: bool, alcohol: bool, drug: bool, support_system: str, living_condition: str) -> str:
"""Collecting patient's social history including:
- Occupation
- smoking or not
- alcohol use or not
- drug use or not
- living conditions
- support system
Returns:
The updated data with the patient's social history added.
"""
@tool
def review_system(weight_change: str, fever: bool, chill: bool, night_sweats: bool, sleep: str, gastrointestinal: str, urinary: str) -> str:
"""Collecting patient's review information including:
- Recent weight changes
- Constitutional symptoms (fever, chills, night sweats)
- Sleep patterns
- Gastrointestinal and urinary function
Returns:
The updated data with the patient's review.
"""
@tool
def confirm_data() -> str:
"""Asks the patient if the data intake is correct.
Returns:
The user's free-text response.
"""
@tool
def get_data() -> str:
"""Returns the users data so far. One item per line."""
@tool
def clear_data():
"""Removes all items from the user's order."""
@tool
def save_data() -> int:
"""Send the data into database.
Returns:
The status of data saving, finished.
"""
def data_node(state: DataState) -> DataState:
"""The ordering node. This is where the dataintake is manipulated."""
tool_msg = state.get("messages", [])[-1]
data = state.get("data", [])
outbound_msgs = []
data_saved = False
for tool_call in tool_msg.tool_calls:
if tool_call["name"] == "patient_id":
# Each order item is just a string. This is where it assembled as "drink (modifiers, ...)".
data["ID"]["name"]=tool_call["args"]["name"]
data["ID"]["DOB"]=tool_call["args"]["DOB"]
data["ID"]["gender"]=tool_call["args"]["gender"]
data["ID"]["contact"]=tool_call["args"]["contact"]
data["ID"]["emergency_contact"]=tool_call["args"]["emergency_contact"]
response = "\n".join(data)
elif tool_call["name"] == "symptom":
# Each order item is just a string. This is where it assembled as "drink (modifiers, ...)".
data["symptom"]["main_symptom"]=tool_call["args"]["main_symptom"]
data["symptom"]["symptom_length"]=tool_call["args"]["length"]
response = "\n".join(data)
elif tool_call["name"] == "pain":
data["pain"]["pain_location"] = tool_call["args"]["pain_location"]
data["pain"]["pain_side"] = tool_call["args"]["pain_side"]
data["pain"]["pain_intensity"] = tool_call["args"]["pain_intensity"]
data["pain"]["pain_description"] = tool_call["args"]["pain_description"]
data["pain"]["start_time"] = tool_call["args"]["start_time"]
data["pain"]["radiation"] = tool_call["args"]["radiation"]
data["pain"]["triggers"] = tool_call["args"]["triggers"]
data["pain"]["symptom"] = tool_call["args"]["symptom"]
response = "\n".join(data)
elif tool_call["name"] == "medical_hist":
data["medical_hist"]["medical_condition"] = tool_call["args"]["medical_condition"]
data["medical_hist"]["first_time"] = tool_call["args"]["first_time"]
data["medical_hist"]["surgery_history"] = tool_call["args"]["surgery_history"]
data["medical_hist"]["medication"] = tool_call["args"]["medication"]
data["medical_hist"]["allergy"] = tool_call["args"]["allergy"]
response = "\n".join(data)
elif tool_call["name"] == "confirm_data":
# We could entrust the LLM to do order confirmation, but it is a good practice to
# show the user the exact data that comprises their order so that what they confirm
# precisely matches the order that goes to the kitchen - avoiding hallucination
# or reality skew.
# In a real scenario, this is where you would connect your POS screen to show the
# order to the user.
print("Your input data:")
if not data:
print(" (no items)")
for data in data:
print(f" {data}")
response = input("Is this correct? ")
elif tool_call["name"] == "get_data":
response = "\n".join(data) if data else "(no data)"
elif tool_call["name"] == "clear_data":
data.clear()
response = None
elif tool_call["name"] == "save_data":
#order_text = "\n".join(order)
print("Saving the data!")
pprint(data)
# TODO(you!): Implement cafe.
data_saved = True
# response = randint(1, 5) # ETA in minutes
else:
raise NotImplementedError(f'Unknown tool call: {tool_call["name"]}')
# Record the tool results as tool messages.
outbound_msgs.append(
ToolMessage(
content=response,
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": outbound_msgs, "data": data, "finished": data_saved} |