آموزش و راهنما

آموزش LangGraph؛ ساخت Agentهای Stateful، Human-in-the-loop و قابل بازیابی با API درواره

آموزش پروژه‌محور LangGraph؛ ساخت Agentهای Stateful و قابل بازیابی با Tool Calling، Checkpoint، حافظه، Human-in-the-loop، FastAPI و API سازگار با OpenAI درواره.

مقدمه

ساخت یک Agent آزمایشی با یک حلقه ساده نسبتاً آسان است. مدل را فراخوانی می‌کنیم، Tool Call را دریافت می‌کنیم، ابزار را اجرا می‌کنیم و نتیجه را دوباره برای مدل می‌فرستیم. اگر همه‌چیز در یک درخواست کوتاه انجام شود، این معماری می‌تواند کافی باشد.

اما مشکلات اصلی زمانی ظاهر می‌شوند که Agent وارد محیط واقعی شود:

کاربر باید یک عملیات را تأیید کند
اجرای Agent چند دقیقه یا چند ساعت طول می‌کشد
سرور هنگام اجرای Task ری‌استارت می‌شود
یک ابزار خارجی موقتاً در دسترس نیست
Agent باید بعداً از همان مرحله ادامه پیدا کند
مسیر اجرا بر اساس وضعیت تغییر می‌کند
چند Branch یا Subtask وجود دارد
State باید میان مراحل مختلف حفظ شود
تاریخچه گفتگو نباید بعد از Restart از بین برود
بعضی عملیات باید توسط انسان ویرایش یا رد شوند
لازم است به یک Checkpoint قبلی برگردیم
چند Agent تخصصی باید با یکدیگر همکاری کنند

در چنین شرایطی، یک while ساده به‌سرعت به مجموعه‌ای پیچیده از شرط‌ها، State، Retry، Queue، Database و Callback تبدیل می‌شود.

LangGraph برای حل همین دسته از مسائل طراحی شده است. این ابزار یک Runtime و Framework سطح پایین برای Orchestration کردن Workflowها و Agentهای Stateful و طولانی‌مدت ارائه می‌کند.

بر اساس مستندات رسمی LangGraph، تمرکز اصلی LangGraph روی قابلیت‌های زیر است:

Durable Execution
Persistence
Human-in-the-loop
Streaming
Short-term Memory
Long-term Memory
Fault Tolerance
Time Travel
Subgraphs

در این مقاله یک Agent پشتیبانی سفارش می‌سازیم که:

وضعیت خود را در Graph State نگه می‌دارد
از API سازگار با OpenAI درواره استفاده می‌کند
Tool Calling انجام می‌دهد
مسیر اجرا را به‌صورت شرطی انتخاب می‌کند
پیش از عملیات حساس متوقف می‌شود
درخواست تأیید، رد یا ویرایش را می‌پذیرد
با Checkpoint از همان مرحله ادامه پیدا می‌کند
تاریخچه Thread را حفظ می‌کند
برای محیط Production از PostgreSQL استفاده می‌کند
از طریق FastAPI در دسترس قرار می‌گیرد

Base URL مورد استفاده در تمام نمونه‌ها:

https://api.darvareh.ir/v1

LangGraph چیست؟

LangGraph یک Runtime برای طراحی و اجرای Workflowها و Agentهای Stateful است. در LangGraph، فرایند را به شکل Graph مدل‌سازی می‌کنیم.

اجزای اصلی Graph عبارت‌اند از:

State: وضعیت مشترک Workflow
Node: یک مرحله پردازش
Edge: مسیر انتقال میان Nodeها
Conditional Edge: انتخاب مسیر بر اساس State
Checkpointer: ذخیره Snapshotهای State
Store: نگهداری حافظه بلندمدت
Interrupt: توقف اجرای Graph برای ورودی خارجی
Command: ادامه یا هدایت اجرای Graph

یک Graph ساده:

START
→ Agent
→ Tools
→ Agent
→ END

Agent Node مدل را فراخوانی می‌کند. اگر مدل Tool Call تولید کند، مسیر به Tools می‌رود. نتیجه ابزار دوباره به Agent داده می‌شود. اگر مدل پاسخ نهایی تولید کند، Graph به END می‌رسد.

LangGraph مدل یا ابزار خاصی را به شما تحمیل نمی‌کند. می‌توانید از LangChain، SDKهای دیگر یا توابع اختصاصی خود استفاده کنید.

تفاوت LangGraph با LangChain

LangChain مجموعه‌ای از Integrationها و Abstractionهای سطح بالاتر برای مدل، Tool، Agent، RAG و Prompt است.

LangGraph روی Orchestration و Runtime تمرکز دارد.

قابلیت	LangChain	LangGraph
اتصال به مدل‌ها	بله	معمولاً از Integrationها استفاده می‌کند
تعریف Tool	بله	ابزار را در Graph اجرا می‌کند
Agent آماده	بله	معماری را خودتان طراحی می‌کنید
StateGraph	محدود	قابلیت اصلی
Persistence	از طریق Runtime زیرین	قابلیت اصلی
Interrupt و Resume	از طریق LangGraph	قابلیت اصلی
Time Travel	از طریق LangGraph	بله
کنترل مسیر اجرا	سطح بالاتر	دقیق و سطح پایین
Durable Execution	از طریق LangGraph	بله
Subgraph	محدود	بله

LangChain Agentها در نسخه‌های جدید روی LangGraph ساخته می‌شوند. اگر یک Agent استاندارد با Tool Loop معمولی می‌خواهید، Abstraction سطح بالاتر LangChain می‌تواند کافی باشد. اگر State، Branch، Checkpoint و کنترل دقیق مسیر لازم دارید، LangGraph انتخاب مناسب‌تری است.

تفاوت LangGraph با OpenAI Agents SDK

OpenAI Agents SDK ساخت Agent با Tools، Handoffs، Guardrails، Sessions و Tracing را ساده می‌کند. LangGraph کنترل دقیق‌تری روی State Machine و اجرای Durable ارائه می‌دهد.

قابلیت	OpenAI Agents SDK	LangGraph
Agent Loop آماده	بله	باید Graph را طراحی کنید
Function Tool	بله	از LangChain Tool یا تابع استفاده می‌کند
Handoff	داخلی	با Node، Edge یا Subgraph
State سفارشی	Context و Session	هسته معماری
Persistence	Session و Integrationهای Durable	Checkpointer
Human-in-the-loop	پشتیبانی می‌شود	Interrupt و Command
Conditional Branch	محدودتر	قابلیت اصلی
Time Travel	وابسته به Runtime	بله
کنترل Graph	کمتر	بسیار زیاد
پیچیدگی شروع	کمتر	بیشتر

اگر هدف ساخت سریع یک Agent استاندارد است، OpenAI Agents SDK ساده‌تر است. اگر Workflow شما باید قابل توقف، قابل بازیابی، شاخه‌ای و دارای State پیچیده باشد، LangGraph مناسب‌تر است.

چه زمانی از LangGraph استفاده کنیم؟

LangGraph برای این سناریوها مناسب است:

Agent طولانی‌مدت
Workflow دارای چند Branch
عملیات نیازمند تأیید انسانی
بازگشت به مرحله قبلی
ذخیره وضعیت میان درخواست‌ها
بازیابی بعد از خطا یا Restart
Multi-Agent System
اجرای موازی چند Subtask
کنترل دقیق Retry
Agent متصل به چند ابزار حساس
Workflowهای مالی یا سازمانی
پردازش اسناد چندمرحله‌ای
Coding Agent
Agent پشتیبانی مشتری
Agent عملیات و Incident Management

LangGraph ممکن است برای این موارد بیش از حد پیچیده باشد:

یک درخواست ساده مدل
Chatbot بدون Tool
RAG یک‌مرحله‌ای
طبقه‌بندی متن
استخراج Structured Output
Workflow کاملاً خطی و کوتاه

اگر مسئله با یک تابع یا Chain ساده حل می‌شود، نیازی به Graph نیست.

معماری پروژه

Agent پشتیبانی این مقاله مسیر زیر را خواهد داشت:

START
→ Agent Node
→ آیا Tool Call وجود دارد؟
   → خیر: END
   → بله: Tool Router
       → ابزار خواندنی: Tool Node
       → ابزار حساس: Approval Node
→ Tool Result
→ Agent Node
→ END

در ابزار حساس:

Approval Node
→ Interrupt
→ انتظار برای تصمیم انسان
   → Approve: Execute Tool
   → Edit: Execute Edited Tool
   → Reject: Return Rejection
→ Agent Node

پیش‌نیازها

Python 3.11 یا جدیدتر
API Key درواره
Model ID دارای Tool Calling
PostgreSQL برای نسخه Production
آشنایی با Python و FastAPI

نصب Dependencyها

pip install -U \
  langgraph \
  langchain-openai \
  langchain-core \
  python-dotenv \
  fastapi \
  uvicorn

برای SQLite Checkpoint:

pip install -U langgraph-checkpoint-sqlite

برای PostgreSQL:

pip install -U \
  langgraph-checkpoint-postgres \
  psycopg[binary,pool]

فایل requirements.txt:

langgraph
langchain-openai
langchain-core
langgraph-checkpoint-sqlite
python-dotenv
fastapi
uvicorn

تنظیم Environment

فایل .env:

DARVAREH_API_KEY=YOUR_DARVAREH_API_KEY
DARVAREH_BASE_URL=https://api.darvareh.ir/v1
DARVAREH_MODEL_ID=YOUR_MODEL_ID
DATABASE_URL=postgresql://postgres:password@localhost:5432/agents

فایل .gitignore:

.env
.venv/
__pycache__/
*.pyc
.pytest_cache/
checkpoints.db

دریافت Model ID از درواره

curl https://api.darvareh.ir/v1/models \
  -H "Authorization: Bearer $DARVAREH_API_KEY"

مقدار id مدل موردنظر را در DARVAREH_MODEL_ID قرار دهید.

برای Agent ابزارمحور، مدل را از نظر موارد زیر آزمایش کنید:

Tool Calling
تولید JSON معتبر
پیروی از Instructions
پایداری در چند Turn
Streaming
Context Window
هزینه و Latency

اتصال ChatOpenAI به درواره

مستندات رسمی ChatOpenAI امکان تعیین base_url و api_key سفارشی را فراهم می‌کنند. مستندات ChatOpenAI

فایل model.py:

import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI


load_dotenv()

model = ChatOpenAI(
    model=os.environ["DARVAREH_MODEL_ID"],
    api_key=os.environ["DARVAREH_API_KEY"],
    base_url=os.getenv(
        "DARVAREH_BASE_URL",
        "https://api.darvareh.ir/v1",
    ),
    temperature=0,
    timeout=30,
    max_retries=2,
)

نکته سازگاری

ChatOpenAI بر اساس فرمت‌های استاندارد OpenAI کار می‌کند. فیلدهای اختصاصی و غیراستاندارد بعضی Providerها ممکن است توسط آن استخراج یا حفظ نشوند.

در این مقاله از قابلیت‌های استاندارد استفاده می‌کنیم:

Chat Completions
Messages
Tool Calling
Streaming استاندارد
Usage Metadata، در صورت ارائه Provider

آزمایش اتصال

from model import model


response = model.invoke(
    "فقط عبارت اتصال برقرار است را بنویس."
)

print(response.content)

نسخه Async:

response = await model.ainvoke(
    "فقط عبارت اتصال برقرار است را بنویس."
)

تعریف ابزارها

فایل tools.py:

from typing import Annotated

from langchain_core.tools import tool


ORDERS = {
    "ORD-1001": {
        "user_id": "user-42",
        "organization_id": "org-1",
        "status": "delayed",
        "tracking_code": "TRK-90001",
        "estimated_delivery": "1405-04-25",
    }
}

TRACKING_TICKETS: dict[str, dict] = {}


@tool
def get_order_status(
    order_id: Annotated[
        str,
        "The unique order identifier.",
    ],
    user_id: Annotated[
        str,
        "The authenticated user identifier.",
    ],
    organization_id: Annotated[
        str,
        "The authenticated organization identifier.",
    ],
) -> dict:
    """Return an order owned by the authenticated user."""

    order = ORDERS.get(order_id)

    if order is None:
        return {
            "success": False,
            "error": "order_not_found",
        }

    if (
        order["user_id"] != user_id
        or order["organization_id"] != organization_id
    ):
        return {
            "success": False,
            "error": "order_not_found_or_access_denied",
        }

    return {
        "success": True,
        "order_id": order_id,
        "status": order["status"],
        "tracking_code": order["tracking_code"],
        "estimated_delivery": order["estimated_delivery"],
    }


@tool
def create_tracking_ticket(
    order_id: Annotated[
        str,
        "The delayed order identifier.",
    ],
    reason: Annotated[
        str,
        "The reason for creating a tracking ticket.",
    ],
    user_id: Annotated[
        str,
        "The authenticated user identifier.",
    ],
    organization_id: Annotated[
        str,
        "The authenticated organization identifier.",
    ],
) -> dict:
    """Create a support ticket for a delayed order."""

    order = ORDERS.get(order_id)

    if order is None:
        return {
            "success": False,
            "error": "order_not_found",
        }

    if (
        order["user_id"] != user_id
        or order["organization_id"] != organization_id
    ):
        return {
            "success": False,
            "error": "order_not_found_or_access_denied",
        }

    if order["status"] not in {
        "delayed",
        "shipped",
    }:
        return {
            "success": False,
            "error": "order_not_eligible",
        }

    idempotency_key = (
        f"{organization_id}:"
        f"{user_id}:"
        f"{order_id}:tracking"
    )

    existing = TRACKING_TICKETS.get(
        idempotency_key
    )

    if existing:
        return {
            "success": True,
            "status": "already_exists",
            **existing,
        }

    ticket = {
        "ticket_id": (
            f"TICKET-{len(TRACKING_TICKETS) + 1}"
        ),
        "order_id": order_id,
        "reason": reason,
    }

    TRACKING_TICKETS[
        idempotency_key
    ] = ticket

    return {
        "success": True,
        "status": "created",
        **ticket,
    }

یک مشکل امنیتی در Schema ابزار

در کد بالا مدل می‌تواند user_id و organization_id تولید کند. حتی اگر Backend آن‌ها را بررسی کند، بهتر است داده هویتی اصلاً در اختیار مدل نباشد.

در نسخه Production:

ابزار عمومی فقط order_id و reason دریافت کند
user_id و organization_id از Runtime Context تزریق شوند
Business Service در لایه پایین‌تر Authorization را انجام دهد

برای ساده نگه داشتن Graph اولیه، فعلاً این فیلدها را در State قرار می‌دهیم و قبل از اجرای ابزار مقدار تولیدشده مدل را با State جایگزین می‌کنیم.

تعریف State

State داده مشترک میان Nodeهای Graph است.

فایل state.py:

from typing import Annotated, TypedDict

from langgraph.graph.message import add_messages


class SupportState(TypedDict):
    messages: Annotated[list, add_messages]

    user_id: str
    organization_id: str

    pending_tool_call: dict | None
    approval_status: str | None

    step_count: int
    error: str | None

Reducer پیام‌ها

این تعریف:

messages: Annotated[list, add_messages]

به LangGraph می‌گوید پیام‌های جدید باید به تاریخچه اضافه شوند، نه اینکه مقدار قبلی کامل جایگزین شود.

بدون Reducer، بازگرداندن:

{
    "messages": [new_message]
}

ممکن است تاریخچه قبلی را جایگزین کند.

تعریف Instructions

فایل prompts.py:

SYSTEM_PROMPT = """
شما عامل پشتیبانی سفارش هستید.

وظایف:
- بررسی وضعیت سفارش
- توضیح وضعیت ارسال
- پیشنهاد ثبت درخواست پیگیری برای سفارش تأخیردار

قواعد:
- اطلاعات سفارش را فقط از ابزار دریافت کنید.
- وضعیت، تاریخ تحویل و کد رهگیری را حدس نزنید.
- برای ثبت درخواست پیگیری از ابزار create_tracking_ticket استفاده کنید.
- اجرای ابزار create_tracking_ticket نیازمند تأیید انسانی است.
- اطلاعات کاربران یا سازمان‌های دیگر را نمایش ندهید.
- بازپرداخت یا لغو سفارش در محدوده شما نیست.
- نتیجه خطای ابزار را شفاف اعلام کنید.
- پاسخ را دقیق و به زبان فارسی ارائه دهید.
"""

متصل کردن ابزارها به مدل

from model import model
from tools import (
    create_tracking_ticket,
    get_order_status,
)


tools = [
    get_order_status,
    create_tracking_ticket,
]

model_with_tools = model.bind_tools(
    tools
)

bind_tools Schema ابزارها را در هر فراخوانی مدل ارسال می‌کند.

اگر تعداد ابزارها بسیار زیاد شود:

Token ورودی افزایش پیدا می‌کند
انتخاب ابزار دشوار می‌شود
احتمال Tool Call اشتباه افزایش می‌یابد

فقط ابزارهای مرتبط با همان Graph یا Agent را Bind کنید.

ساخت Agent Node

فایل nodes.py:

from langchain_core.messages import SystemMessage

from model import model_with_tools
from prompts import SYSTEM_PROMPT
from state import SupportState


async def agent_node(
    state: SupportState,
) -> dict:
    step_count = state.get(
        "step_count",
        0,
    ) + 1

    if step_count > 8:
        return {
            "step_count": step_count,
            "error": "max_steps_exceeded",
        }

    messages = [
        SystemMessage(
            content=SYSTEM_PROMPT
        ),
        *state["messages"],
    ]

    response = await model_with_tools.ainvoke(
        messages
    )

    return {
        "messages": [response],
        "step_count": step_count,
    }

Routing بعد از Agent

اگر آخرین پیام Tool Call داشته باشد، Graph باید به ابزار برود. در غیر این صورت پایان می‌یابد.

from typing import Literal

from langchain_core.messages import AIMessage

from state import SupportState


def route_after_agent(
    state: SupportState,
) -> Literal[
    "prepare_tool",
    "end",
]:
    if state.get("error"):
        return "end"

    last_message = state["messages"][-1]

    if (
        isinstance(last_message, AIMessage)
        and last_message.tool_calls
    ):
        return "prepare_tool"

    return "end"

آماده‌سازی Tool Call

در این Node، داده‌های هویتی امن را از State روی آرگومان‌ها اعمال می‌کنیم.

READ_ONLY_TOOLS = {
    "get_order_status",
}

SENSITIVE_TOOLS = {
    "create_tracking_ticket",
}


def prepare_tool_node(
    state: SupportState,
) -> dict:
    last_message = state["messages"][-1]
    tool_call = last_message.tool_calls[0]

    arguments = dict(
        tool_call.get("args", {})
    )

    arguments["user_id"] = state["user_id"]
    arguments["organization_id"] = (
        state["organization_id"]
    )

    pending = {
        "id": tool_call["id"],
        "name": tool_call["name"],
        "args": arguments,
    }

    return {
        "pending_tool_call": pending,
        "approval_status": None,
    }

برای ساده ماندن آموزش، فقط اولین Tool Call را مدیریت می‌کنیم. در Production باید Tool Callهای موازی یا متعدد را آگاهانه مدیریت کنید.

Routing ابزار

def route_tool(
    state: SupportState,
) -> str:
    pending = state.get(
        "pending_tool_call"
    )

    if not pending:
        return "end"

    tool_name = pending["name"]

    if tool_name in SENSITIVE_TOOLS:
        return "approval"

    if tool_name in READ_ONLY_TOOLS:
        return "execute_tool"

    return "unknown_tool"

اجرای ابزار خواندنی

فهرست ابزارها:

TOOL_MAP = {
    "get_order_status": get_order_status,
    "create_tracking_ticket": (
        create_tracking_ticket
    ),
}

Node اجرا:

from langchain_core.messages import ToolMessage


async def execute_tool_node(
    state: SupportState,
) -> dict:
    pending = state[
        "pending_tool_call"
    ]

    tool = TOOL_MAP.get(
        pending["name"]
    )

    if tool is None:
        result = {
            "success": False,
            "error": "unknown_tool",
        }
    else:
        result = await tool.ainvoke(
            pending["args"]
        )

    return {
        "messages": [
            ToolMessage(
                content=str(result),
                tool_call_id=pending["id"],
            )
        ],
        "pending_tool_call": None,
    }

در نسخه Production بهتر است خروجی ابزار با JSON معتبر serialize شود:

import json

content=json.dumps(
    result,
    ensure_ascii=False,
)

ساخت Approval Node با Interrupt

from langgraph.types import interrupt


def approval_node(
    state: SupportState,
) -> dict:
    pending = state[
        "pending_tool_call"
    ]

    decision = interrupt(
        {
            "type": "tool_approval",
            "tool_name": pending["name"],
            "arguments": pending["args"],
            "allowed_decisions": [
                "approve",
                "edit",
                "reject",
            ],
        }
    )

    return {
        "approval_status": (
            decision.get("decision")
        ),
        "pending_tool_call": (
            decision.get(
                "tool_call",
                pending,
            )
        ),
    }

وقتی اجرای Graph به interrupt() می‌رسد:

State با Checkpointer ذخیره می‌شود
اجرای Graph متوقف می‌شود
Payload تأیید به Caller بازگردانده می‌شود
Graph تا زمان Resume منتظر می‌ماند
هنگام Resume، مقدار ارسال‌شده به Command به‌عنوان خروجی interrupt() برگردانده می‌شود

برای Interrupt به Checkpointer و thread_id نیاز داریم. مستندات Interrupt

Routing بعد از تأیید

def route_after_approval(
    state: SupportState,
) -> str:
    status = state.get(
        "approval_status"
    )

    if status in {
        "approve",
        "edit",
    }:
        return "execute_tool"

    return "reject_tool"

مدیریت رد عملیات

from langchain_core.messages import ToolMessage


def reject_tool_node(
    state: SupportState,
) -> dict:
    pending = state[
        "pending_tool_call"
    ]

    return {
        "messages": [
            ToolMessage(
                content=(
                    "عملیات توسط کاربر رد شد "
                    "و اجرا نشد."
                ),
                tool_call_id=pending["id"],
            )
        ],
        "pending_tool_call": None,
    }

مدل پس از مشاهده این Tool Message می‌تواند پاسخ طبیعی تولید کند.

ابزار ناشناخته

def unknown_tool_node(
    state: SupportState,
) -> dict:
    pending = state[
        "pending_tool_call"
    ]

    return {
        "messages": [
            ToolMessage(
                content=(
                    '{"success":false,'
                    '"error":"unknown_tool"}'
                ),
                tool_call_id=pending["id"],
            )
        ],
        "pending_tool_call": None,
    }

ساخت Graph

فایل graph.py:

from langgraph.graph import (
    END,
    START,
    StateGraph,
)
from langgraph.checkpoint.memory import (
    InMemorySaver,
)

from nodes import (
    agent_node,
    approval_node,
    execute_tool_node,
    prepare_tool_node,
    reject_tool_node,
    route_after_agent,
    route_after_approval,
    route_tool,
    unknown_tool_node,
)
from state import SupportState


builder = StateGraph(
    SupportState
)

builder.add_node(
    "agent",
    agent_node,
)

builder.add_node(
    "prepare_tool",
    prepare_tool_node,
)

builder.add_node(
    "approval",
    approval_node,
)

builder.add_node(
    "execute_tool",
    execute_tool_node,
)

builder.add_node(
    "reject_tool",
    reject_tool_node,
)

builder.add_node(
    "unknown_tool",
    unknown_tool_node,
)

builder.add_edge(
    START,
    "agent",
)

builder.add_conditional_edges(
    "agent",
    route_after_agent,
    {
        "prepare_tool": "prepare_tool",
        "end": END,
    },
)

builder.add_conditional_edges(
    "prepare_tool",
    route_tool,
    {
        "approval": "approval",
        "execute_tool": "execute_tool",
        "unknown_tool": "unknown_tool",
        "end": END,
    },
)

builder.add_conditional_edges(
    "approval",
    route_after_approval,
    {
        "execute_tool": "execute_tool",
        "reject_tool": "reject_tool",
    },
)

builder.add_edge(
    "execute_tool",
    "agent",
)

builder.add_edge(
    "reject_tool",
    "agent",
)

builder.add_edge(
    "unknown_tool",
    "agent",
)

checkpointer = InMemorySaver()

graph = builder.compile(
    checkpointer=checkpointer,
)

InMemorySaver فقط برای توسعه مناسب است. بعد از Restart تمام Checkpointها از بین می‌روند.

اجرای Graph

import asyncio
import uuid

from langchain_core.messages import HumanMessage

from graph import graph


async def main():
    thread_id = str(
        uuid.uuid4()
    )

    config = {
        "configurable": {
            "thread_id": thread_id,
        }
    }

    result = await graph.ainvoke(
        {
            "messages": [
                HumanMessage(
                    content=(
                        "وضعیت سفارش "
                        "ORD-1001 را بررسی کن."
                    )
                )
            ],
            "user_id": "user-42",
            "organization_id": "org-1",
            "pending_tool_call": None,
            "approval_status": None,
            "step_count": 0,
            "error": None,
        },
        config=config,
    )

    print(
        result["messages"][-1].content
    )


asyncio.run(main())

درخواست عملیات حساس

result = await graph.ainvoke(
    {
        "messages": [
            HumanMessage(
                content=(
                    "برای سفارش ORD-1001 "
                    "یک درخواست پیگیری ثبت کن."
                )
            )
        ],
        "user_id": "user-42",
        "organization_id": "org-1",
        "step_count": 0,
    },
    config=config,
)

اگر Graph متوقف شود، در حالت invoke معمولاً اطلاعات Interrupt در __interrupt__ قرار می‌گیرد:

interrupts = result.get(
    "__interrupt__",
    []
)

for item in interrupts:
    print(item.value)

Resume پس از تأیید

باید همان thread_id را استفاده کنیم.

from langgraph.types import Command


resumed = await graph.ainvoke(
    Command(
        resume={
            "decision": "approve",
        }
    ),
    config=config,
)

print(
    resumed["messages"][-1].content
)

اگر thread_id دیگری استفاده شود، LangGraph State متوقف‌شده را پیدا نمی‌کند.

رد عملیات

resumed = await graph.ainvoke(
    Command(
        resume={
            "decision": "reject",
        }
    ),
    config=config,
)

ویرایش Tool Call

resumed = await graph.ainvoke(
    Command(
        resume={
            "decision": "edit",
            "tool_call": {
                "id": pending_id,
                "name": (
                    "create_tracking_ticket"
                ),
                "args": {
                    "order_id": "ORD-1001",
                    "reason": (
                        "تأخیر بیشتر از زمان "
                        "اعلام‌شده"
                    ),
                    "user_id": "user-42",
                    "organization_id": "org-1",
                },
            },
        }
    ),
    config=config,
)

در Production نباید اجازه دهید کاربر با Edit، user_id یا organization_id را تغییر دهد. پس از Resume نیز مقادیر امن را دوباره از Session روی Tool Call اعمال کنید.

قواعد مهم Interrupt

بر اساس مستندات رسمی LangGraph:

Interrupt به Checkpointer نیاز دارد
Resume باید با همان thread_id انجام شود
Payload باید JSON-serializable باشد
کد Node هنگام Resume از ابتدای Node دوباره اجرا می‌شود
Side Effectهای قبل از Interrupt باید Idempotent باشند
ترتیب Interruptها در Node نباید بی‌دلیل تغییر کند
interrupt() را در try/except نامناسب قرار ندهید

مهم‌ترین نکته:

هنگام Resume، Node از ابتدای خود دوباره اجرا می‌شود.

بنابراین این کد خطرناک است:

def approval_node(state):
    create_database_record()
    decision = interrupt("Approve?")

ممکن است create_database_record() هنگام Resume دوباره اجرا شود.

روش صحیح:

def approval_node(state):
    decision = interrupt("Approve?")
    create_database_record()

یا عملیات قبل از Interrupt را Idempotent کنید.

Persistence و Checkpoint

Checkpointer State یک Thread را در هر مرحله ذخیره می‌کند.

کاربردها:

ادامه گفتگو
Human-in-the-loop
Recovery
Time Travel
Debug
مشاهده State قبلی
Resume بعد از Restart

LangGraph میان دو مفهوم تفاوت قائل می‌شود:

مفهوم	کاربرد
Checkpointer	State کوتاه‌مدت یک Thread
Store	حافظه بلندمدت میان Threadها

این تفکیک در مستندات Persistence توضیح داده شده است.

SQLite Checkpointer

برای توسعه محلی:

import sqlite3

from langgraph.checkpoint.sqlite import (
    SqliteSaver,
)


connection = sqlite3.connect(
    "checkpoints.db",
    check_same_thread=False,
)

checkpointer = SqliteSaver(
    connection
)

graph = builder.compile(
    checkpointer=checkpointer,
)

SQLite برای توسعه، آزمایش و استفاده تک‌پردازه مناسب است. برای سرویس پرترافیک از PostgreSQL استفاده کنید.

PostgreSQL Checkpointer

from langgraph.checkpoint.postgres import (
    PostgresSaver,
)


connection_string = (
    "postgresql://postgres:"
    "password@localhost:5432/agents"
)

with PostgresSaver.from_conn_string(
    connection_string
) as checkpointer:
    checkpointer.setup()

    graph = builder.compile(
        checkpointer=checkpointer,
    )

setup() جدول‌ها و Indexهای لازم را ایجاد می‌کند.

در معماری Async از AsyncPostgresSaver استفاده کنید.

نکات PostgreSQL

Connection Pool تعریف کنید
thread_id کوتاه و پایدار باشد
Retention Policy داشته باشید
Checkpointهای قدیمی را Prune کنید
داده حساس را قبل از State حذف کنید
Backup و Recovery را آزمایش کنید
Migration نسخه State را در نظر بگیرید

مستندات توصیه می‌کنند thread_id کمتر از ۲۵۵ کاراکتر باشد. UUID انتخاب مناسبی است.

طراحی Thread ID

Thread ID نباید مستقیماً یک مقدار قابل حدس مانند user-1 باشد.

روش مناسب:

thread_id = str(
    uuid.uuid4()
)

مالکیت Thread را در جدول برنامه ذخیره کنید:

CREATE TABLE agent_threads (
    id UUID PRIMARY KEY,
    user_id UUID NOT NULL,
    organization_id UUID NOT NULL,
    status TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

پیش از Invoke یا Resume:

SELECT id
FROM agent_threads
WHERE id = :thread_id
  AND user_id = :user_id
  AND organization_id = :organization_id;

thread_id مجوز دسترسی نیست.

حافظه بلندمدت با Store

Checkpointer State همان Thread را نگه می‌دارد. برای نگهداری ترجیح کاربر میان Conversationهای مختلف از Store استفاده می‌شود.

from langgraph.store.memory import (
    InMemoryStore,
)


store = InMemoryStore()

graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
)

نمونه حافظه:

namespace = (
    "user_preferences",
    "user-42",
)

store.put(
    namespace,
    "response_language",
    {
        "value": "fa",
        "source": "explicit_user_choice",
    },
)

بازیابی:

memory = store.get(
    namespace,
    "response_language",
)

در Production از Store پایدار استفاده کنید.

چه چیزی را در حافظه بلندمدت ذخیره کنیم؟

موارد مناسب:

زبان ترجیحی
تنظیمات صریح کاربر
Conventionهای سازمان
Workflow انتخاب‌شده
تصمیم‌های تأییدشده و پایدار

موارد نامناسب:

API Key
رمز عبور
Token دسترسی
اطلاعات پرداخت
تمام متن گفتگو
حدس مدل درباره شخصیت کاربر
داده موقت
نتیجه تأییدنشده ابزار
Prompt Injection

Memory Write باید Policy، Scope، TTL و Audit داشته باشد.

Time Travel

Checkpointها امکان مشاهده یا ادامه از State قبلی را فراهم می‌کنند.

کاربرد:

Debug
اصلاح تصمیم اشتباه
اجرای مسیر جایگزین
بازبینی Human-in-the-loop
مقایسه Branchها

اما Time Travel به معنی بازگرداندن Side Effect خارجی نیست.

اگر Agent ایمیل ارسال کرده یا پرداخت انجام داده باشد، برگشت State باعث لغو آن عملیات نمی‌شود.

برای Side Effectها:

Idempotency
Audit Log
Compensation Action
Transaction
Approval
External Operation ID

لازم است.

Retry Policy

Retry را می‌توان در سطح Node یا Tool مدیریت کرد.

خطاهای قابل Retry:

Timeout
خطای شبکه
۵xx
Rate Limit
اختلال موقت Provider

خطاهای غیرقابل Retry:

401
403
Validation Error
Resource not found
Policy Violation
Tool Call غیرمجاز

Retry باید:

محدود باشد
Exponential Backoff داشته باشد
Jitter داشته باشد
Deadline کلی را رعایت کند
هزینه را کنترل کند
در Trace ثبت شود

جلوگیری از Loop

Graph ممکن است میان Agent و Tool بی‌نهایت بچرخد.

راهکارها:

step_count
Recursion Limit
سقف Tool Call
تشخیص فراخوانی تکراری
Deadline
سقف هزینه
محدودیت Handoff
شرط پایان صریح

در Invocation:

config = {
    "configurable": {
        "thread_id": thread_id,
    },
    "recursion_limit": 20,
}

در State نیز step_count را نگه دارید تا پیام خطای مناسب تولید شود.

Tool Call تکراری

در State یک فهرست Hash نگه دارید:

class SupportState(TypedDict):
    messages: Annotated[list, add_messages]
    executed_tool_calls: list[str]

ساخت Hash:

import hashlib
import json


def tool_call_hash(
    name: str,
    arguments: dict,
) -> str:
    payload = json.dumps(
        {
            "name": name,
            "arguments": arguments,
        },
        sort_keys=True,
        ensure_ascii=False,
    )

    return hashlib.sha256(
        payload.encode()
    ).hexdigest()

قبل از اجرا بررسی کنید Hash قبلاً ثبت نشده باشد. برای ابزارهایی که تکرار مشروع دارند، Policy جدا تعریف کنید.

Streaming

LangGraph امکان Stream کردن انواع مختلف اطلاعات را فراهم می‌کند:

پیام‌ها
State Update
State Value
Event
Custom Event
Interrupt

نمونه ساده:

async for chunk in graph.astream(
    input_state,
    config=config,
    stream_mode="updates",
):
    print(chunk)

برای نمایش Tokenهای مدل:

async for message, metadata in graph.astream(
    input_state,
    config=config,
    stream_mode="messages",
):
    if message.content:
        print(
            message.content,
            end="",
            flush=True,
        )

پشتیبانی دقیق Token Streaming و Usage به مدل و Endpoint انتخاب‌شده بستگی دارد.

ساخت API با FastAPI

فایل main.py:

import uuid

from fastapi import (
    FastAPI,
    HTTPException,
)
from langchain_core.messages import (
    HumanMessage,
)
from langgraph.types import Command
from pydantic import BaseModel, Field

from graph import graph


app = FastAPI(
    title="Darvareh LangGraph Agent",
    version="1.0.0",
)


THREAD_OWNERS: dict[str, dict] = {}


class CreateThreadRequest(BaseModel):
    user_id: str
    organization_id: str


class MessageRequest(BaseModel):
    user_id: str
    organization_id: str
    message: str = Field(
        min_length=1,
        max_length=4000,
    )


class ResumeRequest(BaseModel):
    user_id: str
    organization_id: str
    decision: str
    edited_arguments: dict | None = None


def verify_thread(
    thread_id: str,
    user_id: str,
    organization_id: str,
):
    owner = THREAD_OWNERS.get(
        thread_id
    )

    if owner is None:
        raise HTTPException(
            status_code=404,
            detail="Thread not found.",
        )

    if (
        owner["user_id"] != user_id
        or owner["organization_id"]
        != organization_id
    ):
        raise HTTPException(
            status_code=403,
            detail="Access denied.",
        )


@app.get("/health")
async def health():
    return {
        "status": "ok",
    }


@app.post("/threads")
async def create_thread(
    request: CreateThreadRequest,
):
    thread_id = str(
        uuid.uuid4()
    )

    THREAD_OWNERS[thread_id] = {
        "user_id": request.user_id,
        "organization_id": (
            request.organization_id
        ),
    }

    return {
        "thread_id": thread_id,
    }


@app.post(
    "/threads/{thread_id}/messages"
)
async def send_message(
    thread_id: str,
    request: MessageRequest,
):
    verify_thread(
        thread_id=thread_id,
        user_id=request.user_id,
        organization_id=request.organization_id,
    )

    config = {
        "configurable": {
            "thread_id": thread_id,
        },
        "recursion_limit": 20,
    }

    result = await graph.ainvoke(
        {
            "messages": [
                HumanMessage(
                    content=request.message
                )
            ],
            "user_id": request.user_id,
            "organization_id": (
                request.organization_id
            ),
            "step_count": 0,
            "error": None,
        },
        config=config,
    )

    interrupts = result.get(
        "__interrupt__",
        [],
    )

    if interrupts:
        return {
            "status": (
                "approval_required"
            ),
            "interrupt": (
                interrupts[0].value
            ),
        }

    return {
        "status": "completed",
        "message": (
            result["messages"][-1].content
        ),
    }


@app.post(
    "/threads/{thread_id}/resume"
)
async def resume_thread(
    thread_id: str,
    request: ResumeRequest,
):
    verify_thread(
        thread_id=thread_id,
        user_id=request.user_id,
        organization_id=request.organization_id,
    )

    config = {
        "configurable": {
            "thread_id": thread_id,
        },
        "recursion_limit": 20,
    }

    resume_payload = {
        "decision": request.decision,
    }

    if request.edited_arguments:
        resume_payload[
            "edited_arguments"
        ] = request.edited_arguments

    result = await graph.ainvoke(
        Command(
            resume=resume_payload
        ),
        config=config,
    )

    interrupts = result.get(
        "__interrupt__",
        [],
    )

    if interrupts:
        return {
            "status": (
                "approval_required"
            ),
            "interrupt": (
                interrupts[0].value
            ),
        }

    return {
        "status": "completed",
        "message": (
            result["messages"][-1].content
        ),
    }

در Production، THREAD_OWNERS باید جدول دیتابیس باشد. حافظه Process با Restart از بین می‌رود و میان Workerها مشترک نیست.

اجرای FastAPI

uvicorn main:app --reload

Swagger:

http://127.0.0.1:8000/docs

ساخت Thread

curl -X POST \
  http://127.0.0.1:8000/threads \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-42",
    "organization_id": "org-1"
  }'

ارسال پیام

curl -X POST \
  http://127.0.0.1:8000/threads/THREAD_ID/messages \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-42",
    "organization_id": "org-1",
    "message": "وضعیت سفارش ORD-1001 را بررسی کن."
  }'

درخواست پیگیری

curl -X POST \
  http://127.0.0.1:8000/threads/THREAD_ID/messages \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-42",
    "organization_id": "org-1",
    "message": "برای سفارش ORD-1001 درخواست پیگیری ثبت کن."
  }'

پاسخ:

{
  "status": "approval_required",
  "interrupt": {
    "type": "tool_approval",
    "tool_name": "create_tracking_ticket",
    "arguments": {
      "order_id": "ORD-1001",
      "reason": "تأخیر در تحویل"
    },
    "allowed_decisions": [
      "approve",
      "edit",
      "reject"
    ]
  }
}

تأیید و Resume

curl -X POST \
  http://127.0.0.1:8000/threads/THREAD_ID/resume \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user-42",
    "organization_id": "org-1",
    "decision": "approve"
  }'

Restart و بازیابی

اگر از InMemorySaver استفاده کنید، Restart تمام State را حذف می‌کند.

اگر Checkpointer پایدار مانند PostgreSQL داشته باشید:

Graph در Approval متوقف می‌شود
Checkpoint در دیتابیس ذخیره می‌شود
Process می‌تواند متوقف شود
برنامه دوباره اجرا می‌شود
همان Graph با همان Checkpointer ساخته می‌شود
Command(resume=...) با همان thread_id ارسال می‌شود
Workflow از Checkpoint ادامه می‌یابد

این قابلیت تفاوت اصلی یک Agent آزمایشی با Agent Durable است.

Observability

برای هر Run این اطلاعات را ثبت کنید:

Thread ID
Run ID
User ID Hash
Organization ID
Node Name
Model ID
Token
Latency
Tool Call
Tool Result Status
Interrupt
Approval Decision
Retry
Error
Checkpoint
Final Status
Estimated Cost

Payload حساس را کامل ثبت نکنید.

ساختار Event:

{
  "trace_id": "trace-123",
  "thread_id": "thread-42",
  "node": "execute_tool",
  "tool_name": "create_tracking_ticket",
  "status": "success",
  "latency_ms": 85,
  "approval": "approved"
}

می‌توانید از LangSmith یا OpenTelemetry و سیستم Monitoring خودتان استفاده کنید.

تست Nodeها

Nodeها را مستقل تست کنید:

import pytest
from langchain_core.messages import (
    HumanMessage,
)


@pytest.mark.asyncio
async def test_agent_node():
    state = {
        "messages": [
            HumanMessage(
                content=(
                    "وضعیت سفارش "
                    "ORD-1001 را بگو."
                )
            )
        ],
        "user_id": "user-42",
        "organization_id": "org-1",
        "step_count": 0,
        "error": None,
    }

    result = await agent_node(
        state
    )

    assert result["messages"]
    assert result["step_count"] == 1

برای Unit Test نباید API واقعی مدل را فراخوانی کنید. Model را Mock کنید.

تست Graph

سناریوهای مهم:

پاسخ مستقیم بدون ابزار
Tool Call خواندنی
Tool Call حساس
Approve
Reject
Edit
Resume با Thread اشتباه
Tool ناشناخته
Tool Error
Timeout
Loop
Restart
دسترسی Tenant دیگر
Prompt Injection
Tool Call تکراری

Eval Dataset

نمونه:

{
  "name": "tracking_approval",
  "input": "برای سفارش ORD-1001 پیگیری ثبت کن.",
  "expected_path": [
    "agent",
    "prepare_tool",
    "approval"
  ],
  "must_interrupt": true,
  "forbidden_before_approval": [
    "create_tracking_ticket"
  ]
}

Eval فقط پاسخ نهایی را بررسی نمی‌کند. مسیر Graph نیز باید ارزیابی شود.

Multi-Agent با Subgraph

می‌توان برای هر Agent یک Subgraph ساخت:

Triage Graph
→ Order Subgraph
→ Billing Subgraph
→ Technical Subgraph

Subgraph مناسب است اگر:

State تخصصی دارد
ابزارهای متفاوت دارد
Workflow مستقل دارد
قابلیت استفاده مجدد دارد
سطح دسترسی جدا لازم است

اگر Agentها فقط Prompt متفاوت دارند، ساخت چند Subgraph ممکن است غیرضروری باشد.

اجرای موازی

برای Taskهای مستقل می‌توان Branchهای موازی ساخت:

START
├── Search Knowledge
├── Fetch Customer Data
└── Fetch Order Data
        ↓
     Synthesis

اجرای موازی Latency را کاهش می‌دهد، اما باید:

State Merge تعریف شود
خطای هر Branch مدیریت شود
Timeout مستقل وجود داشته باشد
داده تکراری کنترل شود
عملیات Write موازی ایجاد نشود

امنیت Agent

حداقل دسترسی

هر Graph یا Subgraph فقط ابزارهای موردنیاز را دریافت کند.

Tenant Isolation

organization_id را از Session احراز هویت‌شده دریافت کنید و در تمام Queryها اعمال کنید.

Human Approval

برای عملیات Write، مالی، ارسال پیام، حذف و انتشار Approval تعریف کنید.

Idempotency

هر عملیات تغییر‌دهنده باید Idempotency Key داشته باشد.

Prompt Injection

اسناد، پیام کاربر و خروجی ابزار را داده غیرقابل اعتماد تلقی کنید.

Secret Management

API Key را در Environment یا Secret Manager نگه دارید و هرگز وارد State، Prompt، Checkpoint یا Trace نکنید.

Checkpoint Security

Checkpoint ممکن است شامل پیام و نتیجه ابزار باشد. بنابراین:

Encryption at Rest
Retention Policy
Access Control
Tenant Isolation
Redaction
Audit

ضروری است.

خطاهای رایج

استفاده از InMemorySaver در Production

با Restart همه Stateها حذف می‌شوند.

Resume با Thread ID جدید

Graph نمی‌تواند Checkpoint قبلی را پیدا کند.

Side Effect پیش از Interrupt

هنگام Resume ممکن است دوباره اجرا شود.

ذخیره Secret در State

State در Checkpoint ذخیره می‌شود و ممکن است در Log یا Debug دیده شود.

نداشتن Recursion Limit

Agent ممکن است وارد Loop شود.

اعتماد به Model برای Authorization

مجوز باید در Tool و Backend بررسی شود.

استفاده از LangGraph برای Task ساده

Graph پیچیدگی غیرضروری ایجاد می‌کند.

نداشتن Retention Policy

Checkpointها به‌مرور Storage را پر می‌کنند.

Time Travel بدون Compensation

برگشت State، عملیات خارجی را برنمی‌گرداند.

چک‌لیست Production

Graph

State Type-safe است
Nodeها وظیفه محدود دارند
مسیرهای خطا مشخص‌اند
END قابل دسترس است
Recursion Limit وجود دارد
Loop تشخیص داده می‌شود

Persistence

PostgreSQL Checkpointer استفاده می‌شود
Thread Ownership بررسی می‌شود
Retention Policy وجود دارد
Backup فعال است
State Versioning در نظر گرفته شده است

Human-in-the-loop

عملیات حساس Interrupt دارد
Approve، Edit و Reject پشتیبانی می‌شوند
Resume با همان Thread انجام می‌شود
Approval منقضی می‌شود
ابزار پس از Approval دوباره Authorization را بررسی می‌کند

Tools

Read و Write جدا هستند
Idempotency وجود دارد
Timeout تعریف شده است
خروجی ابزار محدود است
خطاها استاندارد هستند

Security

Secret داخل State نیست
Tenant Isolation وجود دارد
Prompt Injection آزمایش شده است
Checkpoint رمزگذاری یا محدود شده است
Audit Log وجود دارد

Observability

Node Transition ثبت می‌شود
Token و Cost ثبت می‌شوند
Tool Latency ثبت می‌شود
Interrupt و Approval ثبت می‌شوند
خطاها دارای Trace ID هستند

پرسش‌های متداول

LangGraph چیست؟

LangGraph یک Runtime سطح پایین برای ساخت Workflowها و Agentهای Stateful، طولانی‌مدت، قابل توقف و قابل بازیابی است.

تفاوت LangGraph با LangChain چیست؟

LangChain Integrationها و Abstractionهای مدل و Agent را ارائه می‌کند. LangGraph روی Orchestration، State، Persistence، Interrupt و Durable Execution تمرکز دارد.

آیا LangGraph فقط با مدل‌های OpenAI کار می‌کند؟

خیر. LangGraph به Provider خاصی محدود نیست. در این مقاله از ChatOpenAI با Base URL سفارشی درواره استفاده کردیم.

Base URL درواره چیست؟

https://api.darvareh.ir/v1

مدل‌های در دسترس را چگونه ببینیم؟

GET https://api.darvareh.ir/v1/models

Checkpointer چیست؟

سیستمی برای ذخیره Snapshotهای State یک Thread است. برای Conversation، Resume، Time Travel و Fault Tolerance استفاده می‌شود.

Store چه تفاوتی با Checkpointer دارد؟

Checkpointer حافظه کوتاه‌مدت و State یک Thread را نگه می‌دارد. Store داده بلندمدت را میان Threadها حفظ می‌کند.

thread_id چیست؟

شناسه‌ای است که Checkpointer با آن State یک Thread را ذخیره و بازیابی می‌کند.

Interrupt چیست؟

مکانیزمی برای توقف Graph و انتظار برای ورودی خارجی مانند تأیید انسان است.

چگونه Graph را ادامه دهیم؟

با همان thread_id و Command(resume=...).

آیا Graph بعد از Restart ادامه پیدا می‌کند؟

اگر از Checkpointer پایدار مانند PostgreSQL استفاده کنید، بله. InMemorySaver بعد از Restart اطلاعات را از دست می‌دهد.

آیا LangGraph حافظه بلندمدت دارد؟

بله، از طریق Store می‌توان حافظه میان Threadها ساخت. باید Policy، Scope و TTL مناسب تعریف شود.

آیا Time Travel عملیات خارجی را برمی‌گرداند؟

خیر. برگشت State به معنی لغو ایمیل، پرداخت یا تغییر دیتابیس نیست. برای آن‌ها Compensation و Idempotency لازم است.

آیا LangGraph برای Multi-Agent مناسب است؟

بله. Agentهای تخصصی را می‌توان به‌صورت Node یا Subgraph مدل‌سازی کرد.

چگونه جلوی Loop را بگیریم؟

از Recursion Limit، Step Count، Deadline، سقف Tool Call و تشخیص فراخوانی تکراری استفاده کنید.

آیا باید از LangSmith استفاده کنیم؟

الزامی نیست. می‌توانید Observability اختصاصی یا OpenTelemetry داشته باشید. LangSmith برای Trace و Eval یکپارچه با LangGraph طراحی شده است.

جمع‌بندی

LangGraph زمانی ارزش واقعی خود را نشان می‌دهد که Agent از یک حلقه کوتاه فراتر می‌رود و باید State، Persistence، Human Approval، Branching و Recovery داشته باشد.

در این مقاله یاد گرفتیم چگونه:

مدل را با ChatOpenAI به API درواره متصل کنیم
State اختصاصی تعریف کنیم
Node و Edge بسازیم
Tool Call را Route کنیم
عملیات حساس را با Interrupt متوقف کنیم
Graph را با Command(resume=...) ادامه دهیم
State را با Checkpointer ذخیره کنیم
حافظه Thread را از حافظه بلندمدت جدا کنیم
FastAPI را به Graph متصل کنیم
Loop، Retry، Security و Observability را مدیریت کنیم

تنظیم اتصال درواره:

model = ChatOpenAI(
    model="YOUR_MODEL_ID",
    api_key="YOUR_DARVAREH_API_KEY",
    base_url="https://api.darvareh.ir/v1",
)

برای محیط Production:

InMemorySaver را با PostgreSQL جایگزین کنید
مالکیت Thread را در دیتابیس بررسی کنید
Secret را داخل State ذخیره نکنید
ابزارهای Write را Idempotent کنید
عملیات حساس را بعد از Interrupt اجرا کنید
Retention Policy برای Checkpointها داشته باشید
Graph را با Eval Dataset واقعی آزمایش کنید

LangGraph بیشترین ارزش را زمانی ایجاد می‌کند که جریان Agent باید قابل مشاهده، قابل کنترل، قابل توقف و قابل بازیابی باشد. برای Taskهای ساده، Chain یا Agent آماده کافی است؛ اما برای Agentهای سازمانی و طولانی‌مدت، Graph می‌تواند مرز میان یک Demo و یک سیستم قابل اعتماد در Production باشد.

مقالات مرتبط پیشنهادی

برای شروع، یک API Key اختصاصی در درواره ایجاد کنید و فهرست مدل‌های فعال حساب را از Endpoint زیر دریافت کنید:

https://api.darvareh.ir/v1/models

سپس ChatOpenAI را با Base URL درواره تنظیم و یک Graph ساده شامل Agent و Tool Node بسازید. پس از آزمایش مسیر پایه، Checkpoint، Interrupt و Human-in-the-loop را اضافه کنید.

Agent را ابتدا با ابزارهای خواندنی اجرا کنید. ابزارهای تغییر‌دهنده را فقط بعد از پیاده‌سازی Authorization، Idempotency، Approval، Audit Log و Eval وارد Graph کنید.