Spaces:
Paused
Paused
| from typing import TypedDict | |
| from app.models import Citation, ComputedFact, EvidenceItem | |
| from app.services.cricket import ReplayData, state_at | |
| from app.services.intent import IntentClassifier | |
| from app.services.gemini import GeminiService | |
| from app.services.rag import CricketRetriever | |
| class MatchLensState(TypedDict, total=False): | |
| match_id: str | |
| ball_index: int | |
| question: str | |
| conversation_id: str | None | |
| intent: str | |
| match_context: dict | |
| computed_statistics: list[ComputedFact] | |
| retrieved_documents: list | |
| evidence: list[EvidenceItem] | |
| final_answer: str | |
| citations: list[dict] | |
| confidence: str | None | |
| warnings: list[str] | |
| def build_workflow(data: ReplayData, gemini: GeminiService, retriever: CricketRetriever, classifier: IntentClassifier): | |
| from langgraph.graph import END, START, StateGraph | |
| graph = StateGraph(MatchLensState) | |
| def load_match_context(state: MatchLensState) -> MatchLensState: | |
| current = state_at(data, state["ball_index"]) | |
| ctx = current.model_dump() | |
| computed = [ | |
| ComputedFact(label="Score", value=f"{current.score}/{current.wickets}"), | |
| ComputedFact(label="Overs", value=current.overs), | |
| ComputedFact(label="Current run rate", value=f"{current.current_run_rate:.2f}"), | |
| ] | |
| if current.target: | |
| computed.append(ComputedFact(label="Target", value=str(current.target))) | |
| if current.required_run_rate is not None: | |
| computed.append(ComputedFact(label="Required run rate", value=f"{current.required_run_rate:.2f}")) | |
| if current.remaining_runs is not None and current.remaining_balls is not None: | |
| computed.append(ComputedFact(label="Equation", value=f"{current.remaining_runs} runs from {current.remaining_balls} balls")) | |
| if current.partnership: | |
| computed.append(ComputedFact(label="Partnership", value=f"{current.partnership.runs} off {current.partnership.balls}b")) | |
| return { | |
| "match_context": ctx, | |
| "computed_statistics": computed, | |
| } | |
| def classify_intent(state: MatchLensState) -> MatchLensState: | |
| return {"intent": classifier.predict(state["question"]).value} | |
| def retrieve_statistics(state: MatchLensState) -> MatchLensState: | |
| context = state["match_context"] | |
| evidence_items = [ | |
| EvidenceItem( | |
| kind="statistic", | |
| title="Verified replay state", | |
| detail=f"{context.get('batting_team', 'Batting team')} {context['score']}/{context['wickets']} after {context['overs']} ov (CRR {context['current_run_rate']:.2f})", | |
| ball_index=state["ball_index"], | |
| ) | |
| ] | |
| if context.get("target"): | |
| req_rate = context.get("required_run_rate") | |
| req_str = f"{req_rate:.2f}" if req_rate is not None else "N/A" | |
| evidence_items.append( | |
| EvidenceItem( | |
| kind="statistic", | |
| title="Target & run chase equation", | |
| detail=f"Target {context['target']}, needing {context.get('remaining_runs')} from {context.get('remaining_balls')} balls. Required rate: {req_str} RPO", | |
| ball_index=state["ball_index"], | |
| ) | |
| ) | |
| if context.get("partnership"): | |
| p = context["partnership"] | |
| evidence_items.append( | |
| EvidenceItem( | |
| kind="statistic", | |
| title=f"Active partnership ({p['batters'][0]} & {p['batters'][1]})", | |
| detail=f"{p['runs']} runs off {p['balls']} balls", | |
| ball_index=state["ball_index"], | |
| ) | |
| ) | |
| if context.get("last_ball"): | |
| lb = context["last_ball"] | |
| evidence_items.append( | |
| EvidenceItem( | |
| kind="ball", | |
| title=f"Delivery {lb.get('over_label', '')} ({lb.get('bowler', '')} to {lb.get('striker', '')})", | |
| detail=lb.get("commentary", "Ball completed"), | |
| ball_index=state["ball_index"], | |
| ) | |
| ) | |
| return {"evidence": evidence_items} | |
| def retrieve_knowledge(state: MatchLensState) -> MatchLensState: | |
| docs = retriever.retrieve(state["question"], limit=3) | |
| return {"retrieved_documents": docs} | |
| def generate_explanation(state: MatchLensState) -> MatchLensState: | |
| documents = state.get("retrieved_documents", []) | |
| context = state["match_context"] | |
| question = state["question"] | |
| intent = state["intent"] | |
| citations = [ | |
| Citation(title=doc.title, source=doc.source_url or "Qdrant knowledge base", excerpt=doc.content, url=doc.source_url) | |
| for doc in documents | |
| ] | |
| # Add rule evidence items to evidence list for the UI | |
| existing_evidence = list(state.get("evidence", [])) | |
| for doc in documents: | |
| existing_evidence.append( | |
| EvidenceItem( | |
| kind="rule", | |
| title=doc.title, | |
| detail=doc.content[:160] + "...", | |
| ball_index=state["ball_index"], | |
| ) | |
| ) | |
| # Build prompt for Gemini | |
| batting = context.get('batting_team', 'Batting side') | |
| bowling = context.get('bowling_team', 'Bowling side') | |
| doc_texts = "\n\n".join([f"Source: {d.title} ({d.source_url})\n{d.content}" for d in documents]) | |
| prompt = f"""You are MatchLens, an expert cricket analytics companion (similar to ESPNcricinfo and Cricbuzz). | |
| Analyze and answer the user's question with precise cricketing insight. | |
| MATCH SITUATION: | |
| - Match: {context.get('match_id')} ({bowling} vs {batting}) | |
| - Score: {batting} {context['score']}/{context['wickets']} in {context['overs']} overs | |
| - Current Run Rate: {context['current_run_rate']:.2f} | |
| - Target: {context.get('target', 'N/A')} | |
| - Required Run Rate: {context.get('required_run_rate', 'N/A')} | |
| - Equation: {context.get('remaining_runs', 'N/A')} runs needed off {context.get('remaining_balls', 'N/A')} balls | |
| - Partnership: {context.get('partnership', {}).get('runs', 0)} runs off {context.get('partnership', {}).get('balls', 0)} balls between {context.get('partnership', {}).get('batters', ['Batter 1', 'Batter 2'])} | |
| - Current Bowler: {context.get('current_bowler', {}).get('name', 'Bowler')} ({context.get('current_bowler', {}).get('wickets', 0)}-{context.get('current_bowler', {}).get('runs', 0)} in {context.get('current_bowler', {}).get('overs', '0')} ov) | |
| RELEVANT CRICKET KNOWLEDGE & RULES RETRIEVED FROM KNOWLEDGE BASE: | |
| {doc_texts if doc_texts else "No specific external rule clause matched."} | |
| QUESTION: | |
| {question} | |
| INSTRUCTIONS: | |
| 1. Provide a direct, authoritative, and engaging analytical response. | |
| 2. Ground all numbers strictly in the verified match situation above. | |
| 3. If the question relates to rules, decisions, DRS, LBW, free hits, or tactics, explain the rule mechanism and directly cite the retrieved knowledge source. | |
| 4. If asked about win probability, analyze the mathematical factors (wickets in hand vs required run rate escalation vs death-overs leverage) rather than declining flatly. | |
| 5. Use markdown formatting with bullet points for readability. Keep it concise (2 to 4 paragraphs or bullet points). | |
| """ | |
| generated = gemini.explain(prompt) | |
| warnings = [] | |
| if not generated: | |
| warnings.append(f"Gemini unavailable ({gemini.error_kind or gemini.status}); deterministic match analysis returned.") | |
| target_info = f" Chasing {context.get('target')}, needing {context.get('remaining_runs')} runs from {context.get('remaining_balls')} balls (RRR {context.get('required_run_rate'):.2f})." if context.get('target') else "" | |
| fallback = f"At over {context['overs']}, {batting} are {context['score']}/{context['wickets']} (CRR {context['current_run_rate']:.2f}).{target_info}" | |
| generated = fallback | |
| return { | |
| "final_answer": generated, | |
| "citations": citations, | |
| "evidence": existing_evidence, | |
| "warnings": warnings, | |
| } | |
| def validate_grounding(state: MatchLensState) -> MatchLensState: | |
| return {"confidence": "high" if not state.get("warnings") else "medium"} | |
| def format_response(state: MatchLensState) -> MatchLensState: | |
| return state | |
| graph.add_node("load_match_context", load_match_context) | |
| graph.add_node("classify_intent", classify_intent) | |
| graph.add_node("retrieve_statistics", retrieve_statistics) | |
| graph.add_node("retrieve_knowledge", retrieve_knowledge) | |
| graph.add_node("generate_explanation", generate_explanation) | |
| graph.add_node("validate_grounding", validate_grounding) | |
| graph.add_node("format_response", format_response) | |
| graph.add_edge(START, "load_match_context") | |
| graph.add_edge("load_match_context", "classify_intent") | |
| graph.add_edge("classify_intent", "retrieve_statistics") | |
| graph.add_edge("retrieve_statistics", "retrieve_knowledge") | |
| graph.add_edge("retrieve_knowledge", "generate_explanation") | |
| graph.add_edge("generate_explanation", "validate_grounding") | |
| graph.add_edge("validate_grounding", "format_response") | |
| graph.add_edge("format_response", END) | |
| return graph.compile() | |