Spaces:
Sleeping
Sleeping
File size: 2,009 Bytes
3295172 | 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 | from __future__ import annotations
from langgraph.graph import StateGraph, END
from agent.state import AgentState
from agent.nodes import (
classify_intent,
retrieve_rag,
extract_lead_fields,
capture_lead,
generate_response,
)
def _route_after_classify(state: AgentState) -> str:
intent = state.get("intent", "unknown")
if intent == "inquiry":
return "retrieve_rag"
if intent == "collecting":
return "extract_lead_fields"
return "generate_response"
def _route_after_extract(state: AgentState) -> str:
if state.get("lead_name") and state.get("lead_email") and state.get("lead_platform"):
return "capture_lead"
return "generate_response"
def build_graph() -> StateGraph:
g = StateGraph(AgentState)
g.add_node("classify_intent", classify_intent)
g.add_node("retrieve_rag", retrieve_rag)
g.add_node("extract_lead_fields", extract_lead_fields)
g.add_node("capture_lead", capture_lead)
g.add_node("generate_response", generate_response)
g.set_entry_point("classify_intent")
g.add_conditional_edges(
"classify_intent",
_route_after_classify,
{
"retrieve_rag": "retrieve_rag",
"extract_lead_fields": "extract_lead_fields",
"generate_response": "generate_response",
},
)
# After RAG retrieval, generate the response
g.add_edge("retrieve_rag", "generate_response")
# After field extraction, either capture or ask for more
g.add_conditional_edges(
"extract_lead_fields",
_route_after_extract,
{
"capture_lead": "capture_lead",
"generate_response": "generate_response",
},
)
# After capture, generate the confirmation message
g.add_edge("capture_lead", "generate_response")
g.add_edge("generate_response", END)
return g.compile()
_app = None
def get_app():
global _app
if _app is None:
_app = build_graph()
return _app
|