Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| import logging | |
| import os | |
| from .budget import BudgetLedger | |
| from .config_loader import load_budget_config, load_cloud_config | |
| from .diagnosis import DiagnosisResult, FollowUpAnswer, RepairDiagnostician | |
| from .retriever import RetrievedContext | |
| from .indexer import make_embedder | |
| from .responder import Responder, Route | |
| from .retriever import RepairRetriever | |
| from .vehicle_config import load_data_paths, load_vehicle_config, vehicle_summary | |
| logger = logging.getLogger(__name__) | |
| class FixSenseRepairAgent: | |
| """RAG + Top3 diagnosis agent with budget tracking.""" | |
| def __init__( | |
| self, | |
| *, | |
| vehicle_config: str = "config/vehicle_f150_2021.yaml", | |
| qdrant_path: str | None = None, | |
| dtc_seed: str = "data/dtc_seed_f150.json", | |
| retrieve_top_k: int = 12, | |
| llm_backend: str = "cloud", | |
| ) -> None: | |
| self.vehicle_config_path = vehicle_config | |
| budget_cfg = load_budget_config() | |
| self.budget = BudgetLedger.load(".fixsense_budget.json", config=budget_cfg) | |
| cloud_cfg = load_cloud_config() | |
| emb = cloud_cfg.get("embedding") or {} | |
| collection = str(emb.get("collection", "fixsense_rag")) | |
| # precedence: explicit arg > FIXSENSE_QDRANT_PATH env > config/cloud.yaml > legacy default | |
| qdrant_path = str( | |
| qdrant_path or os.environ.get("FIXSENSE_QDRANT_PATH") or emb.get("qdrant_path") or ".fixsense_qdrant" | |
| ) | |
| self.qdrant_path = qdrant_path | |
| vehicle = load_vehicle_config(vehicle_config) | |
| self.vehicle = vehicle | |
| self.vehicle_summary = vehicle_summary(vehicle) | |
| self.retriever = RepairRetriever( | |
| embedder=make_embedder(cloud_cfg=cloud_cfg), | |
| qdrant_path=qdrant_path, | |
| collection=collection, | |
| vehicle_filter=vehicle, | |
| dtc_seed_path=dtc_seed, | |
| ) | |
| cloud = cloud_cfg.get("cloud") or {} | |
| self.diagnostician = RepairDiagnostician( | |
| retriever=self.retriever, | |
| retrieve_top_k=retrieve_top_k, | |
| llm_backend=llm_backend, | |
| llm_model=str(cloud.get("text_model", "qwen-plus")), | |
| budget=self.budget, | |
| ) | |
| self.responder = Responder(self.diagnostician) | |
| def route(self, message: str, history: list[dict] | None = None, lang: str = "zh", allow_clarify: bool = True) -> Route | None: | |
| return self.responder.route(message, history or [], self.vehicle_summary, lang, allow_clarify) | |
| def expand_for_guide(self, ctx: RetrievedContext) -> RetrievedContext: | |
| return self.retriever.expand_clip_steps(ctx) | |
| def guide(self, question: str, ctx: RetrievedContext, lang: str = "zh") -> dict: | |
| return self.responder.guide(question, ctx, self.vehicle_summary, lang) | |
| def lookup(self, question: str, ctx: RetrievedContext, lang: str = "zh") -> dict: | |
| return self.responder.lookup(question, ctx, self.vehicle_summary, lang) | |
| def check_in_scope(self, query: str) -> bool: | |
| return self.diagnostician.check_in_scope(query, vehicle_summary=self.vehicle_summary) | |
| def diagnose(self, query: str, *, lang: str = "zh") -> DiagnosisResult: | |
| return self.diagnostician.diagnose(query, vehicle_summary=self.vehicle_summary, lang=lang) | |
| def diagnose_from_retrieval(self, query: str, ctx: RetrievedContext, *, lang: str = "zh") -> DiagnosisResult: | |
| return self.diagnostician.diagnose_from_context( | |
| query, vehicle_summary=self.vehicle_summary, ctx=ctx, lang=lang | |
| ) | |
| def answer_followup( | |
| self, question: str, ctx: RetrievedContext, *, history: list[dict] | None = None, lang: str = "zh" | |
| ) -> FollowUpAnswer: | |
| return self.diagnostician.answer_followup( | |
| question, vehicle_summary=self.vehicle_summary, ctx=ctx, history=history, lang=lang | |
| ) | |
| def rewrite_followup( | |
| self, question: str, original_query: str, history: list[dict] | None = None, lang: str = "zh" | |
| ) -> str: | |
| return self.diagnostician.rewrite_followup(question, original_query, history or [], vehicle_summary=self.vehicle_summary) | |
| def diagnose_json(self, query: str) -> str: | |
| result = self.diagnose(query) | |
| payload = result.model_dump() | |
| payload["budget"] = self.budget.summary() | |
| return json.dumps(payload, ensure_ascii=False, indent=2) | |
| def repl(self) -> None: | |
| print(f"FixSense Agent | {self.vehicle_summary}") | |
| print(f"Budget remaining: ${self.budget.remaining:.2f} / ${self.budget.max_usd:.2f}") | |
| print("Enter fault code or symptom (empty line to quit).\n") | |
| while True: | |
| try: | |
| q = input("技师> ").strip() | |
| except (EOFError, KeyboardInterrupt): | |
| print("\nBye.") | |
| break | |
| if not q: | |
| break | |
| try: | |
| print(self.diagnose_json(q)) | |
| except Exception as e: | |
| print(f"Error: {e}") | |
| print(f"\n[Budget] remaining ${self.budget.remaining:.2f}\n") | |