گرهها، یالها و کامپایل در LangGraph
توسعه عاملهای هوشمند (Agentic Ai) با Langchain و Langgraph
- Create node functions that return valid state updates.
- Connect START, nodes, edges, and END in a StateGraph.
- Compile, visualize, invoke, and inspect a graph.
از State تا Workflow قابل اجرا
**هدف:** ساخت، کامپایل، مشاهده و اجرای یک workflow سهگرهی که `input` را به `output` تبدیل میکند.
**Builds on ← 7.1: LangGraph State and Reducers**
هر node بخشی از `state` را بهروزرسانی میکند و LangGraph این update را با state ترکیب میکند.
```mermaid
flowchart LR
S([START]) --> A[normalize]
A --> B[analyze]
B --> C[format]
C --> E([END])
```
- **node:** تابعی که state را میخواند و یک `dict` از updateها برمیگرداند.
- **edge:** اتصال مسیر اجرا بین دو node.
- **compile:** تبدیل تعریف گراف به runnable graph؛ هنوز اجرا نیست.
**Where this leads → 7.3 Conditional Routing:** edgeها بعداً میتوانند مسیر را بر اساس یک تصمیم انتخاب کنند.
**Where this leads → 7.4 Agent Graph with ToolNode:** nodeها به LLM و ابزار وصل میشوند.
Node معتبر؛ ورودی و خروجی
**قرارداد node**
```python
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
input: str
output: str
def normalize_node(state: State):
text = state["input"]
cleaned = text.strip().lower()
return {"input": cleaned}
```
- ورودی: state فعلی، مانند `state["input"]`
- خروجی معتبر: `dict` با کلیدهای تعریفشده در State
- خروجی، update است؛ نه کل state و نه یک مقدار تنها
**مثال کامل:**
```python
state = {"input": " Hello Graph! "}
text = state["input"]
text = text.strip() # "Hello Graph!"
text = text.lower() # "hello graph!"
update = {"input": text}
```
نتیجه: `{"input": "hello graph!"}`
**چرا؟** LangGraph این update را روی state ادغام میکند؛ پس `return "hello"` قرارداد node را نقض میکند.
سه node؛ هر مرحله یک مسئولیت
**State مشترک**
```python
class State(TypedDict):
input: str
normalized: str
analysis: str
output: str
```
**Nodeها با محاسبه کامل**
```python
def normalize(state: State):
value = state["input"]
value = value.strip()
value = value.lower()
return {"normalized": value}
def analyze(state: State):
value = state["normalized"]
length = len(value)
result = f"length={length}"
return {"analysis": result}
def format_output(state: State):
normalized = state["normalized"]
analysis = state["analysis"]
result = normalized + " | " + analysis
return {"output": result}
```
**اجرای دستی با ورودی مشخص:**
```text
input = " LangGraph "
normalize: strip → "LangGraph"
normalize: lower → "langgraph"
analyze: len("langgraph") → 9
analysis → "length=9"
format: "langgraph" + " | " + "length=9"
output → "langgraph | length=9"
```
هر node فقط update مربوط به خودش را تولید میکند.
```mermaid
flowchart LR
I["input: LangGraph"] --> N["normalize\nnormalized: langgraph"]
N --> A["analyze\nanalysis: length=9"]
A --> F["format\noutput: langgraph | length=9"]
```
اتصال مسیر با START و END
**تعریف StateGraph و edgeها**
```python
builder = StateGraph(State)
builder.add_node("normalize", normalize)
builder.add_node("analyze", analyze)
builder.add_node("format", format_output)
builder.add_edge(START, "normalize")
builder.add_edge("normalize", "analyze")
builder.add_edge("analyze", "format")
builder.add_edge("format", END)
```
```mermaid
flowchart TD
S([START]) --> N[normalize]
N --> A[analyze]
A --> F[format]
F --> E([END])
```
- `START`: نقطه ورود ویژه؛ اجرای گراف از اینجا آغاز میشود.
- `END`: نقطه پایان ویژه؛ اجرای موفق اینجا تمام میشود.
- نام node در edge باید دقیقاً با نام ثبتشده یکی باشد.
**ترجمه هر اتصال:**
```text
START → normalize
normalize → analyze
analyze → format
format → END
```
**مثال وابستگی:** `analyze` مقدار `normalized` را میخواند؛ بنابراین اتصال `normalize → analyze` باید وجود داشته باشد.
Compile، Visualize، Invoke
**سه مرحله متفاوت**
```mermaid
flowchart LR
D["تعریف node و edge"] --> C[compile]
C --> V[visualize]
C --> I[invoke]
I --> U["اجرای واقعی nodeها"]
```
### 1) Compile
```python
graph = builder.compile()
```
`compile()` گراف را بررسی و runnable میسازد؛ nodeها در این خط اجرا نمیشوند.
### 2) Visualize
```python
print(graph.get_graph().draw_ascii())
```
مسیر مورد انتظار: `START → normalize → analyze → format → END`
### 3) Invoke؛ اجرای کامل با همه گامها
```python
initial = {"input": " LangGraph "}
result = graph.invoke(initial)
```
**گامهای state:**
```text
initial: {input: " LangGraph "}
پس از normalize: {input: ..., normalized: "langgraph"}
پس از analyze: {input: ..., normalized: ..., analysis: "length=9"}
پس از format: {input: ..., normalized: ..., analysis: ..., output: "langgraph | length=9"}
```
**نتیجه نهایی:**
```python
result["output"] == "langgraph | length=9"
```
بازبینی نهایی و مسیر بعدی
**چکلیست workflow سهگرهی**
```mermaid
flowchart TD
A["State با کلیدهای لازم"] --> B["nodeهای تابعی"]
B --> C["edge از START تا END"]
C --> D[compile]
D --> E["visualize برای بررسی"]
E --> F["invoke با initial state"]
F --> G["final output"]
```
**شرطهای صحت**
- هر node شکل `state → dict update` دارد.
- هر کلید خواندهشده، از ابتدا در state موجود است یا node قبلی آن را ساخته است.
- مسیر ورود از `START` و مسیر خروج به `END` مشخص است.
- `compile()` اجرا نیست؛ `invoke()` اجراست.
**نمونه نهایی**
```python
graph = builder.compile()
result = graph.invoke({"input": " LangGraph "})
print(result["output"])
# langgraph | length=9
```
**Where this leads → 7.3:** `conditional_edges` مسیر را با تصمیم ساختاری انتخاب میکند.
**Where this leads → 7.4:** همین الگو، اسکلت Agent Graph با LLM و `ToolNode` میشود.
**Where this leads → 7.5:** state و اجرای گراف بعداً با checkpoint و thread identifier قابل پیگیری میشوند.
**Where this leads → 9.4:** همین graph میتواند compiled graph entry point در یک پروژه محلی باشد.
بازگشت به دوره