LangGraph State and Reducers
توسعه عاملهای هوشمند (Agentic Ai) با Langchain و Langgraph
- Define typed graph state schemas.
- Explain state updates, channels, and reducers.
- Implement append semantics for message lists and safe aggregation for parallel outputs.
چرا State مهم است؟
### ایدهٔ اصلی
- در درس 5.1، **Conversation State** تاریخچهٔ کوتاه گفتوگو را نگه میداشت.
- در LangGraph، **State** دادهٔ مشترک میان گرههاست.
- هر گره فقط تغییر موردنیاز را برمیگرداند، نه کل State.
### مثال عددی و دادهای
اگر State فعلی چنین باشد:
\[
S_0=\{\text{messages}: [H("سلام")],\ \text{total}:10\}
\]
و یک Node فقط این Update را برگرداند:
\[
U=\{\text{messages}: [AI("سلام!")],\ \text{total}:3\}
\]
آنگاه Reducerها مقدار جدید هر کلید را میسازند؛ Node مجبور نیست `messages` و `total` کامل را دوباره بنویسد.
```mermaid
flowchart LR
A["State فعلی"] --> B["Node: یک کار"]
B --> C["Update جزئی"]
C --> D["Reducerها"]
D --> E["State جدید"]
```
**Builds on ← Conversation State and Short-Term Memory:** `messages` همان حافظهٔ کوتاهمدت است.
**Where this leads → 7.2 Nodes, Edges, and Compilation:** گرهها باید Update معتبر برای همین State برگردانند.
تعریف Typed State
### State schema
```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
import operator
class State(TypedDict):
messages: Annotated[list, add_messages]
total: Annotated[int, operator.add]
```
- `TypedDict`: شکل و نوع کلیدهای State را مشخص میکند.
- `messages`: فهرست پیامها با قانون ترکیب `add_messages`.
- `total`: عددی که Updateهای آن با جمع ترکیب میشوند.
### چرا `Annotated`؟
نوع داده را نگه میدارد و کنار آن قانون ترکیب را ثبت میکند.
### مثال کوچک
اگر مقدار فعلی `total` برابر `10` و Update برابر `3` باشد:
\[
\operatorname{add}(10,3)=10+3=13
\]
اگر `messages` برابر `[H("سلام")]` و Update برابر `[AI("سلام!")]` باشد، نتیجهٔ معمول:
\[
[H("سلام"),AI("سلام!")]
\]
```mermaid
flowchart TD
S["State"] --> M["messages: list"]
S --> T["total: int"]
M --> R1["add_messages"]
T --> R2["operator.add"]
```
Update و Reducer در عمل
### قرارداد Node
```python
def chatbot(state: State):
response = llm.invoke(state["messages"])
return {"messages": [response]}
```
- ورودی Node: `state` کامل.
- خروجی Node: یک `dict` از Updateهای جزئی.
- Reducer برای هر کلید، مقدار فعلی و Update را ترکیب میکند:
\[
\text{new value}=\text{reducer}(\text{current value},\ \text{node update})
\]
### مثال کامل
مقدار فعلی:
\[
\text{messages}=[H("سلام")],\qquad \text{total}=10
\]
Update گره:
\[
U=\{\text{messages}:[AI("سلام!")],\ \text{total}:3\}
\]
گامهای ترکیب:
\[
\text{messages}_{new}=add\_messages([H("سلام")],[AI("سلام!")])
\]
\[
\text{messages}_{new}=[H("سلام"),AI("سلام!")]
\]
\[
\text{total}_{new}=operator.add(10,3)
\]
\[
\text{total}_{new}=10+3=13
\]
**نتیجه:** Node کل State را برنمیگرداند؛ فقط Update را برمیگرداند.
```tikz
\begin{tikzpicture}[scale=0.8]
\draw[->,gray] (-0.2,0) -- (5.4,0) node[right] {$x$};
\draw[->,gray] (0,-0.2) -- (0,3.6) node[above] {$y$};
\draw[->,blue,thick] (0,0) -- (2.2,1.2) node[above] {$H("سلام")$};
\draw[->,red,thick] (2.2,1.2) -- (4.6,3.0) node[above] {$AI("سلام!")$};
\node[below] at (2.3,-0.35) {$messages$};
\end{tikzpicture}
```
پیامها و Aggregation موازی
### دو Node موازی
```python
class State(TypedDict):
messages: Annotated[list, add_messages]
scores: Annotated[list[int], operator.add]
def worker_a(state: State):
return {"scores": [8]}
def worker_b(state: State):
return {"scores": [5]}
```
فرض: `scores = []`
\[
\text{scores}_1=operator.add([], [8])
\]
\[
\text{scores}_1=[]+[8]=[8]
\]
\[
\text{scores}_2=operator.add([8],[5])
\]
\[
\text{scores}_2=[8]+[5]=[8,5]
\]
```mermaid
flowchart LR
S["State: scores = []"] --> A["worker_a: [8]"]
S --> B["worker_b: [5]"]
A --> R["Reducer: operator.add"]
B --> R
R --> O["scores = ["8, 5"]"]
```
**بدون Reducer:** دو Update همزمان روی یک کلید میتوانند خطای `INVALID_CONCURRENT_GRAPH_UPDATE` بدهند.
**Where this leads → 8.5 Parallel Work and Result Aggregation:** همین الگو خروجی Workerهای موازی را جمع میکند.
مرزها و الگوی نهایی
### انتخاب Reducer
| کانال | Update نمونه | Reducer | نتیجه برای مقدار فعلی مناسب |
|---|---|---|---|
| `messages` | `[AI("پاسخ")]` | `add_messages` | تاریخچه با پیام تازه ادغام میشود |
| `total` | `3` | `operator.add` | مقدار فعلی بهعلاوهٔ 3 |
| بدون Reducer | `3` | overwrite پیشفرض | مقدار تازه جای مقدار قبلی را میگیرد |
### مثال مقایسهای
اگر `x=10` و Update برابر `3` باشد:
\[
\text{با overwrite: }x_{new}=3
\]
\[
\text{با }operator.add:\quad x_{new}=operator.add(10,3)=10+3=13
\]
پس Reducer باید با معنای کانال هماهنگ باشد.
### الگوی کامل
```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
import operator
class State(TypedDict):
messages: Annotated[list, add_messages]
total: Annotated[int, operator.add]
def node(state: State):
new_message = ...
return {"messages": [new_message], "total": 3}
```
- Update باید دیکشنری باشد.
- هر کلید Update باید در schema تعریف شده باشد.
- برای Updateهای موازی روی یک کلید، Reducer مناسب لازم است.
- **Builds on ← 5.1:** پیام تازه به حافظهٔ کوتاهمدت افزوده میشود.
- **Where this leads → 7.2:** همین schema ورودی و خروجی Nodeهای گراف را کنترل میکند.
Back to course