---
name: langgraph-agent
description: >
  بناء AI Agents و Multi-Agent Systems باستخدام LangGraph و LangChain.
  استخدم هذا الـ skill عند: إنشاء agent ذكاء اصطناعي، بناء multi-agent system،
  تصميم workflow بـ LangGraph، إنشاء RAG pipeline، بناء chatbot متقدم،
  أو أي مهمة تتعلق بالوكلاء الذكية. يشمل 18 نمط تصميم للوكلاء مع أمثلة عملية.
---

# LangGraph AI Agent Builder
# بناء وكلاء الذكاء الاصطناعي بـ LangGraph

## Core Architecture

```python
# Standard LangGraph Agent Structure
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated, Sequence
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    current_step: str
    context: dict
    final_answer: str | None
```

## Agentic Design Patterns

### Pattern 1: ReAct Agent (Reasoning + Acting)
```python
from langgraph.prebuilt import create_react_agent

def build_react_agent(tools: list, model_name: str = "claude-sonnet-4-5-20250929"):
    model = ChatAnthropic(model=model_name)
    return create_react_agent(model, tools)
```

### Pattern 2: Plan-and-Execute
```python
class PlanExecuteState(TypedDict):
    input: str
    plan: list[str]
    past_steps: Annotated[list[tuple], operator.add]
    response: str

def build_plan_execute_agent():
    graph = StateGraph(PlanExecuteState)
    
    graph.add_node("planner", planner_node)
    graph.add_node("executor", executor_node)
    graph.add_node("replanner", replanner_node)
    
    graph.set_entry_point("planner")
    graph.add_edge("planner", "executor")
    graph.add_conditional_edges(
        "executor",
        should_replan,
        {"replan": "replanner", "end": END}
    )
    graph.add_edge("replanner", "executor")
    
    return graph.compile()
```

### Pattern 3: Multi-Agent Supervisor
```python
class SupervisorState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_agent: str
    task_complete: bool

def build_supervisor_system():
    """نظام مشرف متعدد الوكلاء"""
    graph = StateGraph(SupervisorState)
    
    # Supervisor decides which agent to call
    graph.add_node("supervisor", supervisor_node)
    
    # Specialized agents
    graph.add_node("researcher", researcher_agent)
    graph.add_node("coder", coder_agent)
    graph.add_node("reviewer", reviewer_agent)
    
    graph.set_entry_point("supervisor")
    
    graph.add_conditional_edges(
        "supervisor",
        route_to_agent,
        {
            "researcher": "researcher",
            "coder": "coder",
            "reviewer": "reviewer",
            "FINISH": END
        }
    )
    
    # All agents report back to supervisor
    for agent in ["researcher", "coder", "reviewer"]:
        graph.add_edge(agent, "supervisor")
    
    return graph.compile()
```

### Pattern 4: RAG Agent with Arabic Support
```python
class ArabicRAGState(TypedDict):
    query: str
    query_ar: str
    documents: list[Document]
    answer: str
    sources: list[str]

def build_arabic_rag_agent(vectorstore):
    """وكيل RAG مع دعم البحث بالعربية"""
    graph = StateGraph(ArabicRAGState)
    
    graph.add_node("normalize_query", normalize_arabic_query)
    graph.add_node("retrieve", retrieve_documents)
    graph.add_node("grade_documents", grade_relevance)
    graph.add_node("generate", generate_answer)
    graph.add_node("hallucination_check", check_hallucination)
    
    graph.set_entry_point("normalize_query")
    graph.add_edge("normalize_query", "retrieve")
    graph.add_edge("retrieve", "grade_documents")
    graph.add_conditional_edges(
        "grade_documents",
        has_relevant_docs,
        {"yes": "generate", "no": "web_search"}
    )
    graph.add_edge("generate", "hallucination_check")
    graph.add_conditional_edges(
        "hallucination_check",
        is_grounded,
        {"yes": END, "no": "generate"}
    )
    
    return graph.compile()

def normalize_arabic_query(state: ArabicRAGState) -> dict:
    """تطبيع النص العربي للبحث"""
    import re
    query = state["query"]
    # Remove tashkeel
    query = re.sub(r'[\u0617-\u061A\u064B-\u0652]', '', query)
    # Normalize alef
    query = re.sub(r'[إأآا]', 'ا', query)
    # Normalize ya
    query = query.replace('ى', 'ي')
    # Normalize ta marbuta
    query = query.replace('ة', 'ه')
    return {"query_ar": query}
```

### Pattern 5: Document Processing Pipeline
```python
class DocProcessState(TypedDict):
    file_path: str
    file_type: str
    extracted_text: str
    structured_data: dict
    classification: str
    status: str

def build_document_pipeline():
    """خط معالجة المستندات الحكومية"""
    graph = StateGraph(DocProcessState)
    
    graph.add_node("classify", classify_document)
    graph.add_node("extract_text", extract_with_ocr)
    graph.add_node("structure", structure_data)
    graph.add_node("validate", validate_data)
    graph.add_node("store", store_in_database)
    
    graph.set_entry_point("classify")
    graph.add_edge("classify", "extract_text")
    graph.add_edge("extract_text", "structure")
    graph.add_edge("structure", "validate")
    graph.add_conditional_edges(
        "validate",
        is_valid,
        {"valid": "store", "invalid": "manual_review"}
    )
    
    return graph.compile()
```

### Pattern 6: Human-in-the-Loop
```python
from langgraph.checkpoint.sqlite import SqliteSaver

def build_hitl_agent():
    """وكيل مع تدخل بشري"""
    graph = StateGraph(AgentState)
    
    graph.add_node("agent", agent_node)
    graph.add_node("human_review", human_review_node)
    graph.add_node("execute", execute_node)
    
    graph.set_entry_point("agent")
    graph.add_conditional_edges(
        "agent",
        needs_approval,
        {"needs_approval": "human_review", "auto_approve": "execute"}
    )
    graph.add_edge("human_review", "execute")
    
    # Enable persistence for interrupt/resume
    memory = SqliteSaver.from_conn_string(":memory:")
    return graph.compile(checkpointer=memory, interrupt_before=["human_review"])
```

## Tools & Integrations

```python
from langchain_core.tools import tool

@tool
def search_government_database(query: str, department: str) -> str:
    """البحث في قاعدة البيانات الحكومية"""
    # Implementation
    pass

@tool  
def generate_government_report(report_type: str, parameters: dict) -> str:
    """إنشاء تقرير حكومي"""
    pass

@tool
def send_notification(user_id: int, message_ar: str, channel: str = "email") -> bool:
    """إرسال إشعار للمستخدم"""
    pass

@tool
def check_compliance(document_id: int) -> dict:
    """فحص الامتثال للوائح"""
    pass
```

## LangSmith Observability

```python
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"
os.environ["LANGCHAIN_PROJECT"] = "gov-platform-agents"

# All LangGraph runs are automatically traced
# Access traces at https://smith.langchain.com
```

## Production Deployment

```python
from langgraph.graph import StateGraph
from langserve import add_routes
from fastapi import FastAPI

app = FastAPI(title="Gov Platform AI Agents")

# Deploy as API endpoint
agent = build_supervisor_system()
add_routes(app, agent, path="/api/agent")

# Health check
@app.get("/health")
async def health():
    return {"status": "healthy", "agents": ["supervisor", "rag", "document"]}
```

## Key Conventions

- Always use TypedDict for state definitions (type safety)
- Implement checkpointing for long-running workflows
- Use structured output for agent responses
- Arabic text normalization before any NLP operation
- Log all agent decisions for audit trail
- Implement retry logic with exponential backoff
- Use streaming for real-time user feedback
- Rate limit external API calls