Spaces:
Sleeping
Sleeping
File size: 11,341 Bytes
90f1f18 c2d68af 90f1f18 3e595a5 90f1f18 3e595a5 90f1f18 62f919c 90f1f18 c2d68af 90f1f18 c2d68af 90f1f18 3e595a5 f52aa71 3e595a5 90f1f18 3e595a5 90f1f18 62f919c 90f1f18 3e595a5 90f1f18 3e595a5 f52aa71 3e595a5 18793f5 3e595a5 62f919c 3e595a5 c2d68af 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 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 |
import pprint
import random
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: str, 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, smoke: 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 pain_manage(pain_medication: str, specialist: bool, other_therapy: str, effectiveness: bool) -> str:
"""Collecting patient's pain management including:
- Current pain medications
- Specialist consultations
- Alternative therapies
- Treatment effectiveness
Returns:
The updated data with the patient's pain management.
"""
@tool
def functional(life_quality: str, limit_activity: str, mood: str) -> str:
"""Collecting patient's functional assement information including:
- Impact on quality of life
- Activity limitations
- Mood and emotional state
Returns:
The updated data with the patient's functional assessment information.
"""
@tool
def plan(goal: str, expectation: str, alternative_treatment: str) -> str:
"""Collecting patient's future treatment plan information including:
- Treatment goals
- Patient expectations
- Alternative treatment considerations
Returns:
The updated data with the patient's future treatment plan information.
"""
@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"]["symptom_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"] == "family_hist":
data["family_hist"]["family_history"] = tool_call["args"]["family_history"]
response = "\n".join(data)
elif tool_call["name"] == "social_hist":
data["social_hist"]["occupation"] = tool_call["args"]["occupation"]
data["social_hist"]["smoke"] = tool_call["args"]["smoke"]
data["social_hist"]["alcohol"] = tool_call["args"]["alcohol"]
data["social_hist"]["drug"] = tool_call["args"]["drug"]
data["social_hist"]["support_system"] = tool_call["args"]["support_system"]
data["social_hist"]["living_condition"] = tool_call["args"]["living_condition"]
response = "\n".join(data)
elif tool_call["name"] == "review_system":
data["review_system"]["weight_change"] = tool_call["args"]["weight_change"]
data["review_system"]["fever"] = tool_call["args"]["fever"]
data["review_system"]["chill"] = tool_call["args"]["chill"]
data["review_system"]["night_sweats"] = tool_call["args"]["night_sweats"]
data["review_system"]["sleep"] = tool_call["args"]["sleep"]
data["review_system"]["gastrointestinal"] = tool_call["args"]["gastrointestinal"]
data["review_system"]["urinary"] = tool_call["args"]["urinary"]
response = "\n".join(data)
elif tool_call["name"] == "pain_manage":
data["pain_manage"]["pain_medication"] = tool_call["args"]["pain_medication"]
data["pain_manage"]["specialist"] = tool_call["args"]["specialist"]
data["pain_manage"]["other_therapy"] = tool_call["args"]["other_therapy"]
data["pain_manage"]["effectiveness"] = tool_call["args"]["effectiveness"]
response = "\n".join(data)
elif tool_call["name"] == "functional":
data["functional"]["life_quality"] = tool_call["args"]["life_quality"]
data["functional"]["limit_activity"] = tool_call["args"]["limit_activity"]
data["functional"]["mood"] = tool_call["args"]["mood"]
response = "\n".join(data)
elif tool_call["name"] == "plan":
data["plan"]["goal"] = tool_call["args"]["goal"]
data["plan"]["expectation"] = tool_call["args"]["expectation"]
data["plan"]["alternative_treatment"] = tool_call["args"]["alternative_treatment"]
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)")
print(state["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!")
print(data)
# TODO(you!): Implement cafe.
data_saved = True
response = random.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} |