ماشین‌های وضعیت LangGraph: مدیریت جریان وظایف پیچیده عامل در تولید

Summarize this content to 400 words in Persian Lang

LangGraph چیست؟

LangGraph یک چارچوب ارکستراسیون گردش کار است که به طور خاص برای برنامه های LLM طراحی شده است. اصول اصلی آن عبارتند از:

تقسیم وظایف پیچیده به حالت ها و انتقال ها
مدیریت منطق انتقال حالت
مدیریت استثناهای مختلف در طول اجرای کار

به خرید فکر کنید: مرور → افزودن به سبد خرید → پرداخت → پرداخت. LangGraph به ما کمک می کند تا چنین جریان های کاری را به طور موثر مدیریت کنیم.

مفاهیم اصلی

1. ایالات

ایالت ها مانند ایست های بازرسی در اجرای وظایف شما هستند:

from typing import TypedDict, List

class ShoppingState(TypedDict):
# Current state
current_step: str
# Cart items
cart_items: List[str] # Total amount
total_amount: float
# User input
user_input: str

class 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)

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

2. انتقال دولت

انتقال حالت “نقشه راه” جریان وظیفه شما را تعریف می کند:

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()

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

3. تداوم دولت

برای اطمینان از قابلیت اطمینان سیستم، باید اطلاعات وضعیت را حفظ کنیم:

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

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

4. مکانیسم بازیابی خطا

هر مرحله ای ممکن است شکست بخورد، و ما باید این موقعیت ها را با ظرافت مدیریت کنیم:

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, State

class CustomerServiceState(TypedDict):
conversation_history: List[str] current_intent: str
user_info: dict
resolved: bool

class 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 system
graph = 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
)

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

بهترین شیوه ها

اصول طراحی دولتی

حالت ها را ساده و واضح نگه دارید
فقط اطلاعات لازم را ذخیره کنید
الزامات سریال سازی را در نظر بگیرید

بهینه سازی منطق انتقال

از انتقال های شرطی استفاده کنید
از حلقه های بی نهایت اجتناب کنید
حداکثر محدودیت گام را تنظیم کنید

استراتژی رسیدگی به خطا

تنزل برازنده را اجرا کنید
ثبت اطلاعات دقیق
مکانیزم‌های برگشت را ارائه دهید

بهینه سازی عملکرد

از عملیات ناهمزمان استفاده کنید
اجرای ذخیره سازی حالت
کنترل اندازه حالت

مشکلات و راه حل های رایج

انفجار دولتی

مشکل: بسیاری از ایالت ها تعمیر و نگهداری را دشوار می کنند
راه حل: حالت های مشابه را ادغام کنید، به جای ایجاد حالت های جدید از ترکیب های حالت استفاده کنید

موقعیت های بن بست

مشکل: انتقال حالت دایره ای باعث معلق ماندن وظایف می شود
راه حل: مکانیسم های وقفه و شرایط خروج اجباری را اضافه کنید

قوام دولت

مشکل: حالت های ناسازگار در محیط های توزیع شده
راه حل: از قفل های توزیع شده و مکانیسم های تراکنش استفاده کنید

خلاصه

ماشین‌های حالت LangGraph راه‌حلی قدرتمند برای مدیریت جریان‌های پیچیده AI Agent ارائه می‌دهند:

مدیریت جریان کار را پاک کنید
تداوم وضعیت قابل اعتماد
مدیریت جامع خطا
قابلیت توسعه انعطاف پذیر

from typing import TypedDict, List

class ShoppingState(TypedDict):
    # Current state
    current_step: str
    # Cart items
    cart_items: List[str]
    # Total amount
    total_amount: float
    # User input
    user_input: str

class 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, State

class CustomerServiceState(TypedDict):
    conversation_history: List[str]
    current_intent: str
    user_info: dict
    resolved: bool

class 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 system
graph = 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
        )