LangGraph 状态机：管理生产中的复杂代理任务流

from typing import typeddict, listclass shoppingstate(typeddict):    # current state    current_step: str    # cart items    cart_items: list[str]    # total amount    total_amount: float    # user input    user_input: strclass shoppinggraph(stategraph):    def __init__(self):        super().__init__()        # define states        self.add_node("browse", self.browse_products)        self.add_node("add_to_cart", self.add_to_cart)        self.add_node("checkout", self.checkout)        self.add_node("payment", self.payment)

class shoppingcontroller:    def define_transitions(self):        # add transition rules        self.graph.add_edge("browse", "add_to_cart")        self.graph.add_edge("add_to_cart", "browse")        self.graph.add_edge("add_to_cart", "checkout")        self.graph.add_edge("checkout", "payment")    def should_move_to_cart(self, state: shoppingstate) -> bool:        """determine if we should transition to cart state"""        return "add to cart" in state["user_input"].lower()

class statemanager:    def __init__(self):        self.redis_client = redis.redis()    def save_state(self, session_id: str, state: dict):        """save state to redis"""        self.redis_client.set(            f"shopping_state:{session_id}",            json.dumps(state),            ex=3600  # 1 hour expiration        )    def load_state(self, session_id: str) -> dict:        """load state from redis"""        state_data = self.redis_client.get(f"shopping_state:{session_id}")        return json.loads(state_data) if state_data else none

class errorhandler:    def __init__(self):        self.max_retries = 3    async def with_retry(self, func, state: dict):        """function execution with retry mechanism"""        retries = 0        while retries < self.max_retries:            try:                return await func(state)            except exception as e:                retries += 1                if retries == self.max_retries:                    return self.handle_final_error(e, state)                await self.handle_retry(e, state, retries)    def handle_final_error(self, error, state: dict):        """handle final error"""        # save error state        state["error"] = str(error)        # rollback to last stable state        return self.rollback_to_last_stable_state(state)

from langgraph.graph import stategraph, stateclass customerservicestate(typeddict):    conversation_history: list[str]    current_intent: str    user_info: dict    resolved: boolclass customerservicegraph(stategraph):    def __init__(self):        super().__init__()        # initialize states        self.add_node("greeting", self.greet_customer)        self.add_node("understand_intent", self.analyze_intent)        self.add_node("handle_query", self.process_query)        self.add_node("confirm_resolution", self.check_resolution)    async def greet_customer(self, state: state):        """greet customer"""        response = await self.llm.generate(            prompt=f"""            conversation history: {state['conversation_history']}            task: generate appropriate greeting            requirements:            1. maintain professional friendliness            2. acknowledge returning customers            3. ask how to help            """        )        state['conversation_history'].append(f"assistant: {response}")        return state    async def analyze_intent(self, state: state):        """understand user intent"""        response = await self.llm.generate(            prompt=f"""            conversation history: {state['conversation_history']}            task: analyze user intent            output format:            {{                "intent": "refund/inquiry/complaint/other",                "confidence": 0.95,                "details": "specific description"            }}            """        )        state['current_intent'] = json.loads(response)        return state

# Initialize systemgraph = CustomerServiceGraph()state_manager = StateManager()error_handler = ErrorHandler()async def handle_customer_query(user_id: str, message: str):    # Load or create state    state = state_manager.load_state(user_id) or {        "conversation_history": [],        "current_intent": None,        "user_info": {},        "resolved": False    }    # Add user message    state["conversation_history"].append(f"User: {message}")    # Execute state machine flow    try:        result = await graph.run(state)        # Save state        state_manager.save_state(user_id, result)        return result["conversation_history"][-1]    except Exception as e:        return await error_handler.with_retry(            graph.run,            state        )