1105 / tab1_contact_tracing.py
smartTranscend's picture
Rename tab1_contact_tracing (1).py to tab1_contact_tracing.py
5e6ee0f verified
import streamlit as st
import plotly.graph_objects as go
import numpy as np
import pandas as pd
from openai import OpenAI
# 初始化 tab1 專屬的 session state
def init_tab1_session_state():
if 'tab1_chat_history' not in st.session_state:
st.session_state.tab1_chat_history = []
if 'tab1_selected_ct' not in st.session_state:
st.session_state.tab1_selected_ct = 18.0
if 'tab1_tracing_days' not in st.session_state:
st.session_state.tab1_tracing_days = 7
# 效益計算函數(保持原樣)
def get_effectiveness_for_ct18(days):
if days <= 1: return 0.06
elif days <= 2: return 0.18
elif days <= 3: return 0.31
elif days <= 4: return 0.42
elif days <= 5: return 0.52
elif days <= 6: return 0.60
elif days <= 7: return 0.67
elif days <= 8: return 0.72
elif days <= 9: return 0.77
elif days <= 10: return 0.81
elif days <= 11: return 0.84
elif days <= 12: return 0.87
elif days <= 13: return 0.89
elif days <= 14: return 0.91
elif days <= 15: return 0.92
elif days <= 16: return 0.93
elif days <= 17: return 0.95
elif days <= 20: return 0.97
elif days <= 34: return 0.97 + (days - 20) * 0.03 / 14
else: return 1.0
def get_effectiveness_for_ct22(days):
if days <= 1: return 0.01
elif days <= 2: return 0.03
elif days <= 3: return 0.07
elif days <= 4: return 0.12
elif days <= 5: return 0.18
elif days <= 6: return 0.24
elif days <= 7: return 0.30
elif days <= 8: return 0.36
elif days <= 9: return 0.41
elif days <= 10: return 0.47
elif days <= 11: return 0.52
elif days <= 12: return 0.57
elif days <= 13: return 0.62
elif days <= 14: return 0.66
elif days <= 17: return 0.66 + (days - 14) * 0.10 / 3
elif days <= 21: return 0.76 + (days - 17) * 0.10 / 4
elif days <= 28: return 0.86 + (days - 21) * 0.08 / 7
elif days <= 48: return 0.94 + (days - 28) * 0.06 / 20
else: return 1.0
def get_effectiveness_for_ct30(days):
if days <= 8: return 0
elif days <= 9: return 0.01
elif days <= 10: return 0.02
elif days <= 11: return 0.03
elif days <= 12: return 0.05
elif days <= 13: return 0.08
elif days <= 14: return 0.11
elif days <= 15: return 0.15
elif days <= 16: return 0.20
elif days <= 17: return 0.26
elif days <= 18: return 0.32
elif days <= 19: return 0.37
elif days <= 20: return 0.43
elif days <= 21: return 0.48
elif days <= 24: return 0.48 + (days - 21) * 0.14 / 3
elif days <= 28: return 0.62 + (days - 24) * 0.14 / 4
elif days <= 54: return 0.76 + (days - 28) * 0.24 / 26
else: return 1.0
def get_effectiveness(ct_value, days):
if ct_value <= 18:
ratio = (ct_value - 10) / 8
higher_eff = get_effectiveness_for_ct18(days) * 1.1
effectiveness = higher_eff * (1 - ratio) + get_effectiveness_for_ct18(days) * ratio
elif ct_value <= 22:
ratio = (ct_value - 18) / 4
effectiveness = get_effectiveness_for_ct18(days) * (1 - ratio) + get_effectiveness_for_ct22(days) * ratio
elif ct_value <= 30:
ratio = (ct_value - 22) / 8
effectiveness = get_effectiveness_for_ct22(days) * (1 - ratio) + get_effectiveness_for_ct30(days) * ratio
else:
ratio = min((ct_value - 30) / 5, 1)
effectiveness = get_effectiveness_for_ct30(days) * (1 - ratio * 0.3)
return min(max(effectiveness, 0), 1.0)
def find_days_for_target_effectiveness(ct_value, target_effectiveness):
for days in range(1, 61):
eff = get_effectiveness(ct_value, days)
if eff >= target_effectiveness:
return days
return 60
def get_virus_level(ct_value):
if ct_value <= 18:
return "高病毒量"
elif ct_value <= 25:
return "中病毒量"
else:
return "低病毒量"
def generate_summary_prompt(ct_value, days, effectiveness):
virus_level = get_virus_level(ct_value)
comparison_data = []
for test_days in [3, 7, 14, 21, 30]:
eff = get_effectiveness(ct_value, test_days)
comparison_data.append(f"追蹤{test_days}天: {eff*100:.1f}%")
prompt = f"""你是一位 COVID-19 防疫專家。請根據以下數據生成一份專業的防疫策略效益分析報告:
**當前參數:**
- Ct 值:{ct_value}
- 病毒量等級:{virus_level}
- 回溯追蹤天數:{days}
- 追蹤效益:{effectiveness*100:.1f}%
**不同追蹤天數的效益比較:**
{chr(10).join(comparison_data)}
請提供:
1. 當前策略的效益評估(是否充足?)
2. 基於 Ct 值的感染階段判斷
3. 具體的追蹤建議(是否需要調整天數?)
4. 可能遺漏的接觸者風險評估
5. 實務操作建議
請用繁體中文回答,使用清晰的結構和專業但易懂的語言。"""
return prompt
def extract_query_params(text):
import re
text_lower = text.lower()
# 提取 Ct 值
ct_patterns = [
r'ct\s*[=值]?\s*(\d+\.?\d*)',
r'(\d+\.?\d*)\s*ct',
]
ct_value = None
for pattern in ct_patterns:
match = re.search(pattern, text_lower)
if match:
try:
ct_value = float(match.group(1))
if 10 <= ct_value <= 35:
break
except:
continue
# 提取天數
day_patterns = [
r'(\d+)\s*天',
r'追[蹤跡]\s*(\d+)',
r'回[朔溯]\s*(\d+)',
]
days = None
for pattern in day_patterns:
match = re.search(pattern, text)
if match:
try:
days = int(match.group(1))
if 1 <= days <= 60:
break
except:
continue
# 檢查是否在詢問「要幾天才能達到X%效益」
target_eff = None
reverse_query_patterns = [
r'要.*?(\d+).*?%',
r'達到.*?(\d+).*?%',
r'(\d+).*?%.*?效益',
r'100%',
r'百分之百',
]
for pattern in reverse_query_patterns:
match = re.search(pattern, text)
if match:
if pattern in ['100%', '百分之百']:
target_eff = 1.0
break
else:
try:
target_eff = float(match.group(1)) / 100
if 0 <= target_eff <= 1:
break
except:
continue
return ct_value, days, target_eff
def call_gpt4(prompt, api_key, tools=None, tool_choice=None):
try:
client = OpenAI(api_key=api_key)
messages = [
{"role": "system", "content": "你是一位專業的 COVID-19 防疫策略分析專家,擅長解釋接觸者追蹤效益,並提供清晰實用的建議。當用戶詢問特定 Ct 值和追蹤天數的效益時,你必須使用 calculate_effectiveness 函數來獲取精確數據。當用戶詢問需要多少天才能達到某個效益時,你必須使用 find_required_days 函數。"},
{"role": "user", "content": prompt}
]
if tools:
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
tools=tools,
tool_choice=tool_choice,
temperature=0.7,
max_tokens=1500
)
else:
response = client.chat.completions.create(
model="gpt-4",
messages=messages,
temperature=0.7,
max_tokens=1500
)
return response
except Exception as e:
return f"❌ API 呼叫失敗: {str(e)}"
def process_gpt4_response(response, api_key):
if isinstance(response, str):
return response, None, None
message = response.choices[0].message
if message.tool_calls:
import json
tool_call = message.tool_calls[0]
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result_content = ""
extracted_ct = None
extracted_days = None
if function_name == "calculate_effectiveness":
ct_value = arguments.get("ct_value")
days = arguments.get("days")
extracted_ct = ct_value
extracted_days = days
effectiveness = get_effectiveness(ct_value, days)
virus_level = get_virus_level(ct_value)
result_content = json.dumps({
"ct_value": ct_value,
"days": days,
"effectiveness_percentage": round(effectiveness * 100, 1),
"virus_level": virus_level
}, ensure_ascii=False)
elif function_name == "find_required_days":
ct_value = arguments.get("ct_value")
target_effectiveness = arguments.get("target_effectiveness")
required_days = find_days_for_target_effectiveness(ct_value, target_effectiveness)
actual_effectiveness = get_effectiveness(ct_value, required_days)
virus_level = get_virus_level(ct_value)
extracted_ct = ct_value
extracted_days = required_days
result_content = json.dumps({
"ct_value": ct_value,
"target_effectiveness_percentage": round(target_effectiveness * 100, 1),
"required_days": required_days,
"actual_effectiveness_percentage": round(actual_effectiveness * 100, 1),
"virus_level": virus_level
}, ensure_ascii=False)
client = OpenAI(api_key=api_key)
second_response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一位專業的 COVID-19 防疫策略分析專家,擅長解釋接觸者追蹤效益,並提供清晰實用的建議。"},
{"role": "user", "content": message.content if hasattr(message, 'content') and message.content else "請分析效益"},
{
"role": "assistant",
"content": None,
"tool_calls": [tool_call.model_dump()]
},
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": result_content
}
],
temperature=0.7,
max_tokens=1500
)
return second_response.choices[0].message.content, extracted_ct, extracted_days
return message.content, None, None
def render():
"""渲染 Tab1 的完整內容"""
init_tab1_session_state()
# 側邊欄 - 控制面板
with st.sidebar:
st.header("📊 防疫決策工具 (接觸者追蹤)")
# 互動式參數調整
st.subheader("🎛️ 參數設定")
selected_ct = st.slider(
"🦠 確診者Ct值",
min_value=10.0,
max_value=35.0,
value=st.session_state.tab1_selected_ct,
step=0.5,
help="Ct值越低代表病毒量越高",
key="tab1_ct_slider"
)
st.session_state.tab1_selected_ct = selected_ct
st.caption("10 (高病毒量) ← → 35 (低病毒量)")
tracing_days = st.slider(
"📅 接觸者往回追蹤天數",
min_value=1,
max_value=54,
value=st.session_state.tab1_tracing_days,
help="需要追溯過去幾天的接觸者",
key="tab1_days_slider"
)
st.session_state.tab1_tracing_days = tracing_days
st.caption("1天 ← → 54天")
# 當前效益顯示
current_eff = get_effectiveness(selected_ct, tracing_days)
st.metric(
label="📊 當前情境效益",
value=f"{current_eff * 100:.0f}%",
delta="追蹤效益"
)
# AI 分析按鈕
st.markdown("---")
if st.button("🤖 生成 AI 效益分析報告", type="primary", use_container_width=True, key="tab1_ai_report"):
if not st.session_state.api_key:
st.error("❌ 請先輸入 OpenAI API Key")
else:
with st.spinner("AI 正在分析中..."):
prompt = generate_summary_prompt(selected_ct, tracing_days, current_eff)
response = call_gpt4(prompt, st.session_state.api_key)
final_response, _, _ = process_gpt4_response(response, st.session_state.api_key)
st.session_state.tab1_chat_history.append({
"type": "auto_summary",
"ct": selected_ct,
"days": tracing_days,
"effectiveness": current_eff,
"response": final_response
})
st.markdown("---")
# 情境說明
with st.expander("🎯 使用情境", expanded=False):
st.markdown("""
當發現確診者時,防疫人員需要決定:
**「要往回追蹤多少天的接觸者?」**
""")
with st.expander("✅ 關鍵指標:Ct值"):
st.markdown("""
RT-PCR檢測的「循環閾值」
- **Ct值 ≤ 18** = 病毒量很高
- **Ct值 18-25** = 病毒量中等
- **Ct值 > 25** = 病毒量較低
""")
with st.expander("📈 決策邏輯"):
st.markdown("""
**低Ct值**(高病毒量)
→ 最近才感染
→ 追蹤「近期」接觸者即可
**高Ct值**(低病毒量)
→ 可能感染較久
→ 需追溯「更早」的接觸者
""")
# 主要內容區 - 使用兩欄布局
col1, col2 = st.columns([2, 1])
with col1:
# 生成3D數據
ct_range = np.arange(10, 35.5, 0.5)
day_range = np.arange(1, 61, 1)
ct_mesh, day_mesh = np.meshgrid(ct_range, day_range)
# 計算效益值
effectiveness_mesh = np.zeros_like(ct_mesh)
for i in range(len(day_range)):
for j in range(len(ct_range)):
effectiveness_mesh[i, j] = get_effectiveness(ct_mesh[i, j], day_mesh[i, j])
# 創建3D曲面圖
fig = go.Figure()
# 添加曲面
fig.add_trace(go.Surface(
x=ct_mesh,
y=day_mesh,
z=effectiveness_mesh,
colorscale=[
[0.0, 'rgb(239, 68, 68)'],
[0.33, 'rgb(245, 158, 11)'],
[0.67, 'rgb(16, 185, 129)'],
[1.0, 'rgb(59, 130, 246)']
],
showscale=True,
colorbar=dict(
title="追蹤效益",
tickvals=[0, 0.25, 0.5, 0.75, 1.0],
ticktext=['0%', '25%', '50%', '75%', '100%']
),
opacity=0.9,
name='效益曲面',
contours=dict(
x=dict(show=True, color='white', width=1, highlightwidth=2),
y=dict(show=True, color='white', width=1, highlightwidth=2),
z=dict(show=True, color='white', width=1, highlightwidth=2)
)
))
# 添加當前天數的切片線
ct_line = np.arange(10, 35.1, 0.3)
eff_line = [get_effectiveness(ct, tracing_days) for ct in ct_line]
day_line = [tracing_days] * len(ct_line)
fig.add_trace(go.Scatter3d(
x=ct_line,
y=day_line,
z=eff_line,
mode='lines',
line=dict(color='yellow', width=8),
name=f'第{tracing_days}天'
))
# 添加用戶選擇的點
user_eff = get_effectiveness(selected_ct, tracing_days)
fig.add_trace(go.Scatter3d(
x=[selected_ct],
y=[tracing_days],
z=[user_eff],
mode='markers+text',
marker=dict(size=10, color='blue', symbol='circle',
line=dict(color='white', width=2)),
text=[f'{user_eff*100:.0f}%'],
textposition='top center',
textfont=dict(size=14, color='white'),
name='當前情境'
))
# 設置佈局
fig.update_layout(
title={
'text': '精準追蹤接觸者效益 3D 圖 (Alpha 變異株)',
'x': 0.5,
'xanchor': 'center'
},
scene=dict(
xaxis=dict(
title='X-病毒量(Ct值)',
range=[10, 35],
tickvals=[10, 15, 20, 25, 30, 35],
showgrid=True,
gridwidth=2,
gridcolor='rgb(200, 200, 200)',
showbackground=True,
backgroundcolor='rgba(240, 240, 240, 0.9)'
),
yaxis=dict(
title='Y-回朔追蹤接觸者天數',
range=[1, 60],
tickvals=[1, 10, 20, 30, 40, 50, 60],
showgrid=True,
gridwidth=2,
gridcolor='rgb(200, 200, 200)',
showbackground=True,
backgroundcolor='rgba(240, 240, 240, 0.9)'
),
zaxis=dict(
title='Z-效益',
range=[0, 1],
tickvals=[0, 0.25, 0.5, 0.75, 1.0],
ticktext=['0%', '25%', '50%', '75%', '100%'],
showgrid=True,
gridwidth=2,
gridcolor='rgb(200, 200, 200)',
showbackground=True,
backgroundcolor='rgba(240, 240, 240, 0.9)'
),
camera=dict(
eye=dict(x=1.5, y=-1.5, z=1.3),
center=dict(x=0, y=0, z=0)
),
aspectmode='manual',
aspectratio=dict(x=1, y=1.2, z=0.8)
),
height=700,
showlegend=True,
legend=dict(
x=0.02,
y=0.98,
bgcolor='rgba(255, 255, 255, 0.8)'
),
margin=dict(l=0, r=0, t=40, b=0)
)
st.plotly_chart(fig, use_container_width=True)
# 情境狀態顯示
if selected_ct <= 18:
status = "⚠️ 高病毒量 - 近期感染"
color = "red"
elif selected_ct <= 25:
status = "⚡ 中病毒量 - 感染數天"
color = "orange"
else:
status = "💊 低病毒量 - 感染較久"
color = "blue"
st.info(f"**情境:** 確診者 Ct值 = **{selected_ct}**,追蹤 **{tracing_days}** 天 | {status}")
# 效益解釋
effectiveness = get_effectiveness(selected_ct, tracing_days)
if effectiveness >= 0.9:
st.success("💡 **意義:** 追蹤時間充足,可以找到大部分接觸者")
elif effectiveness >= 0.7:
st.warning("💡 **意義:** 追蹤效果良好,但仍可能遺漏部分接觸者")
elif effectiveness >= 0.5:
st.warning("💡 **意義:** 追蹤效果一般,建議延長追蹤時間")
else:
st.error("💡 **意義:** 追蹤效果不足,需要大幅延長追蹤時間")
with col2:
st.subheader("🤖 AI 助手對話")
# 聊天輸入區
st.markdown("##### 💬 快速查詢")
user_input = st.text_input(
"輸入問題(例如:Ct值20追蹤14天的效益如何?)",
placeholder="例如: Ct 20, 要回溯幾天才能達到100%?",
label_visibility="collapsed",
key="tab1_user_input"
)
if st.button("📤 發送", use_container_width=True, key="tab1_send"):
if not st.session_state.api_key:
st.error("❌ 請先在側邊欄輸入 OpenAI API Key")
elif user_input.strip():
with st.spinner("AI 思考中..."):
# 提取參數
extracted_ct, extracted_days, target_eff = extract_query_params(user_input)
# 構建對話 prompt
chat_prompt = f"""用戶問題:{user_input}
當前系統參數作為參考:
- Ct 值:{selected_ct}
- 追蹤天數:{tracing_days}
- 當前效益:{current_eff*100:.1f}%
請根據用戶的具體問題,使用相應的工具函數來獲取精確數據並給出專業建議。
請用繁體中文回答,簡潔專業。只需要:
1. 直接回答數據結果(Ct值、天數、效益)
2. 提供2-3點簡短的策略調整建議即可
不要解釋接觸者追蹤的背景知識或一般性原理。"""
# 定義 function calling tools
tools = [
{
"type": "function",
"function": {
"name": "calculate_effectiveness",
"description": "計算特定 Ct 值和追蹤天數組合的接觸者追蹤效益。當用戶詢問具體的 Ct 值和天數時,必須使用此函數獲取精確數據。",
"parameters": {
"type": "object",
"properties": {
"ct_value": {
"type": "number",
"description": "確診者的 Ct 值,範圍 10-35。Ct 值越低代表病毒量越高。"
},
"days": {
"type": "integer",
"description": "需要回溯追蹤的天數,範圍 1-60 天。"
}
},
"required": ["ct_value", "days"]
}
}
},
{
"type": "function",
"function": {
"name": "find_required_days",
"description": "反向查詢:給定 Ct 值和目標效益百分比,計算需要回溯多少天才能達到該效益。當用戶詢問「要幾天才能達到X%效益」時使用。",
"parameters": {
"type": "object",
"properties": {
"ct_value": {
"type": "number",
"description": "確診者的 Ct 值,範圍 10-35。"
},
"target_effectiveness": {
"type": "number",
"description": "目標效益,範圍 0-1 (例如 0.9 代表 90%, 1.0 代表 100%)。"
}
},
"required": ["ct_value", "target_effectiveness"]
}
}
}
]
# 根據問題類型決定 tool_choice
tool_choice = "auto"
if target_eff is not None and extracted_ct is not None:
tool_choice = {"type": "function", "function": {"name": "find_required_days"}}
elif extracted_ct and extracted_days:
tool_choice = {"type": "function", "function": {"name": "calculate_effectiveness"}}
response = call_gpt4(chat_prompt, st.session_state.api_key, tools=tools, tool_choice=tool_choice)
final_response, result_ct, result_days = process_gpt4_response(response, st.session_state.api_key)
# 更新參數
if result_ct is not None and result_days is not None:
st.session_state.tab1_selected_ct = float(result_ct)
st.session_state.tab1_tracing_days = int(result_days)
st.session_state.tab1_chat_history.append({
"type": "user_query",
"question": user_input,
"response": final_response
})
# 如果有更新參數,重新渲染頁面
if result_ct is not None and result_days is not None:
st.rerun()
# 顯示對話歷史
st.markdown("---")
st.markdown("##### 📜 對話記錄")
if st.session_state.tab1_chat_history:
# 反向顯示(最新的在上面)
for i, chat in enumerate(reversed(st.session_state.tab1_chat_history)):
with st.container():
if chat["type"] == "auto_summary":
st.markdown(f"**🤖 AI 報告** (Ct={chat['ct']}, {chat['days']}天, 效益={chat['effectiveness']*100:.0f}%)")
st.markdown(chat["response"])
else:
st.markdown(f"**👤 問題:** {chat['question']}")
st.markdown(f"**🤖 回答:** {chat['response']}")
st.markdown("---")
if st.button("🗑️ 清除對話記錄", use_container_width=True, key="tab1_clear_chat"):
st.session_state.tab1_chat_history = []
st.rerun()
else:
st.info("💡 點擊「生成 AI 報告」或在上方輸入問題開始對話")
# 底部說明區域
st.markdown("---")
col_a, col_b = st.columns(2)
with col_a:
with st.expander("💡 操作說明", expanded=False):
st.markdown("""
**3D 圖表操作:**
- 🖱️ **拖曳旋轉**:用滑鼠拖曳3D圖表可以旋轉視角
- 🔍 **縮放**:使用滑鼠滾輪可以縮放
- 🎚️ **調整參數**:使用左側滑桿調整Ct值和追蹤天數
- 💡 **黃色線**:顯示當前追蹤天數下,不同Ct值的效益
- 🔵 **藍色點**:您當前選擇的情境
**AI 助手功能:**
- 📊 **自動報告**:點擊「生成 AI 報告」獲得當前策略的完整分析
- 💬 **快速查詢**:直接輸入 Ct 值和天數快速獲得效益評估
- 🔄 **反向查詢**:詢問「Ct X 要幾天達到 Y% 效益」獲得所需天數
- 🔗 **參數連動**:對話查詢後左側滑桿會自動更新到對應數值
""")
with col_b:
with st.expander("📊 查看詳細數據", expanded=False):
# 創建示例數據表
data = {
'Ct值': [15, 18, 22, 28, 33],
f'追蹤{tracing_days}天的效益': [
f"{get_effectiveness(15, tracing_days)*100:.1f}%",
f"{get_effectiveness(18, tracing_days)*100:.1f}%",
f"{get_effectiveness(22, tracing_days)*100:.1f}%",
f"{get_effectiveness(28, tracing_days)*100:.1f}%",
f"{get_effectiveness(33, tracing_days)*100:.1f}%"
],
'病毒量等級': ['高', '高', '中', '低', '低']
}
df = pd.DataFrame(data)
st.dataframe(df, use_container_width=True)