效果圖
分析流程
代碼實現
廢話少說,直接上代碼
from langchain_core.language_models.llms import BaseLLM
from langchain_core.outputs import Generation, LLMResult
from pydantic.v1 import Field, validator
from typing import Any, Dict, List, Optional, AsyncIterator
import requests
import osclass DeepSeekLLM(BaseLLM):api_key: str = Field(alias="api_key")model: str = "deepseek-chat"temperature: float = 0.7max_tokens: int = 1000# 必須實現的抽象方法def _generate(self,prompts: List[str],stop: Optional[List[str]] = None,**kwargs: Any,) -> LLMResult:print("_generate:")generations = []for prompt in prompts:response = self._call_api(prompt)generations.append([Generation(text=response)])return LLMResult(generations=generations)async def _agenerate(self,prompts: List[str],stop: Optional[List[str]] = None,**kwargs: Any,) -> LLMResult:# 異步實現(可選)return self._generate(prompts, stop, **kwargs)def _call_api(self, prompt: str) -> str:try:headers = {"Authorization": f"Bearer {self.api_key}","Content-Type": "application/json"}payload = {"messages": [{"role": "user", "content": prompt}],"model": self.model,"temperature": self.temperature,"max_tokens": self.max_tokens}#將輸入 輸出都保存到文件中 import json# 添加一個分隔符 沒有 json.txt 就創建with open("json.txt", "a", encoding="utf-8") as f:json.dump(payload, f, ensure_ascii=False, indent=4)response = requests.post("https://api.deepseek.com/v1/chat/completions",headers=headers,json=payload,timeout=30)with open("json.txt", "a", encoding="utf-8") as f:json.dump(response.json(), f, ensure_ascii=False, indent=4)# 增加響應內容驗證if not response.text.strip():raise ValueError("API返回空響應")try:data = response.json()except json.JSONDecodeError:# 嘗試提取可能的JSON片段import rejson_match = re.search(r'```json\n({.*?})\n```', response.text, re.DOTALL)if json_match:data = json.loads(json_match.group(1))else:raise ValueError(f"無法解析API響應: {response.text[:200]}...")# 驗證關鍵字段if not data.get("choices") or not isinstance(data["choices"], list):raise ValueError("API返回格式異常,缺少choices字段")content = data["choices"][0]["message"]["content"]# 清理響應內容(去除可能的Markdown標記)if content.startswith("```json"):content = content[7:-3].strip()return contentexcept requests.exceptions.RequestException as e:raise ValueError(f"API請求失敗: {str(e)}")except Exception as e:raise ValueError(f"處理API響應時出錯: {str(e)}")@propertydef _llm_type(self) -> str:return "deepseek"
car_diagnosis_app.py 代碼?
import streamlit as st
from langchain.chains import SequentialChain, TransformChain
from langchain.memory import ConversationBufferMemory
from langchain.prompts import PromptTemplate
from langchain.chains.llm import LLMChain
from deepseek_llm import DeepSeekLLM
import json
import datetime
import random
import os# 初始化DeepSeek LLM
api_key = os.getenv("DEEPSEEK_API_KEY", st.secrets.get("DEEPSEEK_API_KEY", "your-api-key"))
llm = DeepSeekLLM(api_key=api_key, temperature=0.3, max_tokens=1500)# 模擬車輛數據庫
VEHICLE_DB = {"京A12345": {"brand": "Toyota","model": "Camry","year": 2020,"mileage": 45000,"last_service": "2023-12-15","insurance_expiry": "2024-09-30","next_maintenance": "2024-06-30"},"滬B67890": {"brand": "Honda","model": "CR-V","year": 2019,"mileage": 60000,"last_service": "2023-11-20","insurance_expiry": "2024-08-15","next_maintenance": "2024-05-20"}
}# 模擬維修店數據庫
REPAIR_SHOPS = [{"id": 1, "name": "誠信汽修", "distance": "1.2km", "rating": 4.8, "services": ["機油更換", "剎車維修", "發動機診斷"], "price_level": "$$"},{"id": 2, "name": "途虎養車工場店", "distance": "2.3km", "rating": 4.7, "services": ["輪胎更換", "保養套餐", "電子診斷"], "price_level": "$$$"},{"id": 3, "name": "小李快修", "distance": "0.8km", "rating": 4.5, "services": ["快速保養", "玻璃水加注", "簡單維修"], "price_level": "$"}
]# 模擬知識庫音頻
SOUND_LIBRARY = {"噠噠聲": "https://example.com/sounds/ticking.mp3","嗡嗡聲": "https://example.com/sounds/humming.mp3","吱吱聲": "https://example.com/sounds/squeaking.mp3","咔嗒聲": "https://example.com/sounds/clicking.mp3"
}# 1. 信息提取鏈
def extract_vehicle_info(inputs: dict) -> dict:license_plate = inputs["license_plate"]vehicle_info = VEHICLE_DB.get(license_plate, {})if not vehicle_info:st.error(f"未找到車牌號 {license_plate} 的車輛信息")return {"vehicle_info": "未知車輛"}# 檢查服務提醒today = datetime.date.today()reminders = []insurance_expiry = datetime.datetime.strptime(vehicle_info["insurance_expiry"], "%Y-%m-%d").date()if (insurance_expiry - today).days < 30:reminders.append({"type": "insurance", "message": "您的車輛保險即將到期"})next_maintenance = datetime.datetime.strptime(vehicle_info["next_maintenance"], "%Y-%m-%d").date()if (next_maintenance - today).days < 30:reminders.append({"type": "maintenance", "message": "您的車輛即將需要保養"})return {"vehicle_info": json.dumps(vehicle_info),"reminders": json.dumps(reminders)}info_extraction_chain = TransformChain(input_variables=["license_plate"],output_variables=["vehicle_info", "reminders"],transform=extract_vehicle_info
)# 2. 診斷鏈
def setup_diagnosis_chain():diagnosis_template = """您是一名專業的汽車維修技師,正在幫助車主診斷車輛問題。請根據以下信息進行診斷:車輛信息:{vehicle_info}用戶描述的癥狀:{symptoms}請根據您的專業知識:1. 分析可能的故障原因2. 生成最多3個關鍵問題來進一步明確問題3. 對于異響類問題,請建議用戶試聽哪種聲音樣本輸出格式:{{"analysis": "對問題的初步分析","questions": ["問題1", "問題2", "問題3"],"sound_suggestion": "建議試聽的聲音類型"}}"""prompt = PromptTemplate(template=diagnosis_template,input_variables=["vehicle_info", "symptoms"])return LLMChain(llm=llm, prompt=prompt, output_key="diagnosis_result")# 3. 維修決策鏈
def setup_repair_decision_chain():decision_template = """基于以下診斷信息:{diagnosis_result}請判斷:1. 車主是否能夠自行修復問題?(是/否)2. 如果可自行修復,提供詳細的步驟指導3. 如果需要專業維修,推薦維修項目輸出格式:{{"self_repairable": true/false,"repair_steps": ["步驟1", "步驟2", ...],"recommended_services": ["服務1", "服務2", ...]}}"""prompt = PromptTemplate(template=decision_template,input_variables=["diagnosis_result"])return LLMChain(llm=llm, prompt=prompt, output_key="repair_decision")# 4. 門店推薦鏈
def recommend_shops(inputs: dict) -> dict:decision = json.loads(inputs["repair_decision"])if decision.get("self_repairable", False):return {"shop_recommendations": json.dumps([])}# 根據位置和推薦服務篩選門店location = inputs["location"]recommended_services = decision.get("recommended_services", [])filtered_shops = []for shop in REPAIR_SHOPS:# 簡單匹配服務(實際應用中應有更復雜的匹配邏輯)if any(service in shop["services"] for service in recommended_services):shop["match_score"] = random.uniform(0.7, 1.0) # 模擬匹配度計算filtered_shops.append(shop)# 按距離和評分排序filtered_shops.sort(key=lambda x: (x["distance"], -x["rating"]))return {"shop_recommendations": json.dumps(filtered_shops[:3])}shop_recommendation_chain = TransformChain(input_variables=["repair_decision", "location"],output_variables=["shop_recommendations"],transform=recommend_shops
)# 完整工作流
def create_full_workflow():diagnosis_chain = setup_diagnosis_chain()repair_decision_chain = setup_repair_decision_chain()return SequentialChain(chains=[info_extraction_chain,diagnosis_chain,repair_decision_chain,shop_recommendation_chain],input_variables=["license_plate", "symptoms", "location"],output_variables=["vehicle_info", "reminders", "diagnosis_result", "repair_decision", "shop_recommendations"],verbose=True)# Streamlit UI
def main():st.set_page_config(page_title="智能汽車故障診斷", layout="wide")st.title("🚗 智能汽車故障診斷助手")# 初始化session狀態if "diagnosis_stage" not in st.session_state:st.session_state.diagnosis_stage = "initial"st.session_state.memory = ConversationBufferMemory()st.session_state.workflow = create_full_workflow()st.session_state.answers = {}st.session_state.current_questions = []# 側邊欄 - 車輛信息輸入with st.sidebar:st.header("車輛信息")license_plate = st.text_input("車牌號", "京A12345")location = st.text_input("當前位置", "北京市海淀區")st.header("車輛狀態")if license_plate in VEHICLE_DB:vehicle = VEHICLE_DB[license_plate]st.write(f"品牌: {vehicle['brand']}")st.write(f"型號: {vehicle['model']}")st.write(f"里程: {vehicle['mileage']}公里")st.write(f"上次保養: {vehicle['last_service']}")# 服務提醒today = datetime.date.today()insurance_expiry = datetime.datetime.strptime(vehicle["insurance_expiry"], "%Y-%m-%d").date()if (insurance_expiry - today).days < 30:st.warning(f"? 保險將于 {vehicle['insurance_expiry']} 到期")next_maintenance = datetime.datetime.strptime(vehicle["next_maintenance"], "%Y-%m-%d").date()if (next_maintenance - today).days < 30:st.warning(f"🔧 下次保養時間: {vehicle['next_maintenance']}")# 主界面 - 診斷流程if st.session_state.diagnosis_stage == "initial":st.subheader("請描述您的車輛問題")symptoms = st.text_area("例如:冷啟動時有噠噠異響,儀表盤機油燈閃爍", height=150)if st.button("開始診斷"):if not symptoms.strip():st.error("請輸入車輛問題描述")returnst.session_state.symptoms = symptomsst.session_state.license_plate = license_platest.session_state.location = locationst.session_state.diagnosis_stage = "processing"st.rerun()# 診斷處理中elif st.session_state.diagnosis_stage == "processing":with st.spinner("正在分析您的車輛問題..."):try:# 執行工作流result = st.session_state.workflow({"license_plate": st.session_state.license_plate,"symptoms": st.session_state.symptoms,"location": st.session_state.location})# 解析結果diagnosis_result = json.loads(result["diagnosis_result"])repair_decision = json.loads(result["repair_decision"])shop_recommendations = json.loads(result["shop_recommendations"])# 存儲結果st.session_state.diagnosis_result = diagnosis_resultst.session_state.repair_decision = repair_decisionst.session_state.shop_recommendations = shop_recommendationsst.session_state.current_questions = diagnosis_result.get("questions", [])# 如果有聲音建議,準備音頻sound_type = diagnosis_result.get("sound_suggestion")if sound_type and sound_type in SOUND_LIBRARY:st.session_state.sound_url = SOUND_LIBRARY[sound_type]else:st.session_state.sound_url = Nonest.session_state.diagnosis_stage = "show_results"st.rerun()except Exception as e:st.error(f"診斷過程中發生錯誤: {str(e)}")st.session_state.diagnosis_stage = "initial"# 顯示診斷結果elif st.session_state.diagnosis_stage == "show_results":st.subheader("診斷結果")# 顯示初步分析st.markdown(f"**問題分析:** {st.session_state.diagnosis_result.get('analysis', '')}")# 播放聲音建議if st.session_state.sound_url:st.markdown("**聲音對比:** 請聽以下聲音是否與您的車輛聲音相似")st.audio(st.session_state.sound_url, format='audio/mp3')# 顯示進一步問題if st.session_state.current_questions:st.markdown("**請回答以下問題以進一步明確問題:**")for i, question in enumerate(st.session_state.current_questions):st.session_state.answers[i] = st.text_input(question, key=f"q_{i}")if st.button("提交答案并繼續診斷"):# 將答案添加到癥狀描述中new_symptoms = "\n".join([f"Q: {st.session_state.current_questions[i]}\nA: {st.session_state.answers[i]}" for i in range(len(st.session_state.current_questions))])st.session_state.symptoms += "\n" + new_symptomsst.session_state.diagnosis_stage = "processing"st.rerun()# 顯示維修決策st.divider()decision = st.session_state.repair_decisionif decision.get("self_repairable", False):st.success("? 您可以嘗試自行修復此問題")st.markdown("**修復步驟:**")for i, step in enumerate(decision.get("repair_steps", [])):st.markdown(f"{i+1}. {step}")# 添加AR指導按鈕if st.button("查看AR修復指導"):st.session_state.show_ar = Trueif st.session_state.get("show_ar", False):st.video("https://example.com/ar_repair_guide.mp4")else:st.warning("?? 建議到專業維修店處理此問題")st.markdown(f"**推薦維修項目:** {', '.join(decision.get('recommended_services', []))}")# 顯示推薦門店st.subheader("推薦維修店")if st.session_state.shop_recommendations:cols = st.columns(len(st.session_state.shop_recommendations))for i, shop in enumerate(st.session_state.shop_recommendations):with cols[i]:st.markdown(f"**{shop['name']}**")st.caption(f"距離: {shop['distance']} | 評分: {shop['rating']}")st.caption(f"服務: {', '.join(shop['services'][:3])}")st.caption(f"價格: {shop['price_level']}")if st.button("選擇此門店", key=f"shop_{i}"):st.session_state.selected_shop = shopst.session_state.diagnosis_stage = "shop_selected"st.rerun()else:st.info("未找到匹配的維修店,請嘗試擴大搜索范圍")# 返回按鈕if st.button("重新診斷"):st.session_state.diagnosis_stage = "initial"st.rerun()# 門店選擇后elif st.session_state.diagnosis_stage == "shop_selected":shop = st.session_state.selected_shopst.success(f"您已選擇: {shop['name']}")# 顯示預約信息st.subheader("預約信息")date = st.date_input("預約日期", min_value=datetime.date.today())time = st.time_input("預約時間", datetime.time(10, 00))contact = st.text_input("聯系電話")# 維修項目確認st.subheader("維修項目確認")services = st.session_state.repair_decision.get("recommended_services", [])selected_services = st.multiselect("請確認維修項目", services, default=services)if st.button("確認預約"):st.session_state.appointment = {"shop": shop["name"],"date": date.strftime("%Y-%m-%d"),"time": time.strftime("%H:%M"),"services": selected_services,"contact": contact}st.session_state.diagnosis_stage = "appointment_confirmed"st.rerun()# 預約確認elif st.session_state.diagnosis_stage == "appointment_confirmed":appt = st.session_state.appointmentst.balloons()st.success("🎉 預約成功!")st.markdown(f"""**維修店:** {appt['shop']} **時間:** {appt['date']} {appt['time']} **維修項目:** {', '.join(appt['services'])} **聯系電話:** {appt['contact']}""")st.info("維修店將很快聯系您確認預約詳情")if st.button("返回主頁"):st.session_state.diagnosis_stage = "initial"st.rerun()if __name__ == "__main__":main()
執行腳本
#環境比變量
export DEEPSEEK_API_KEY=your_api_key
#安裝依賴
pip install streamlit langchain requests
#運行應用
streamlit run car_diagnosis_app.py
#訪問應用
http://localhost:8501