ساخت AI Agent با پایتون با LangGraph از صفر تا اجرا

اگر می خواهید یک دستیار هوشمند بسازید که بتواند گفتگو کند، ابزار اجرا کند و وضعیت خود را مدیریت کند، این راهنما مسیر عملی ساخت AI Agent با پایتون و چارچوب LangGraph را با نمونه کد کامل پوشش می دهد. در پایان می توانید یک عامل مبتنی بر الگوی گراف بسازید که درخواست کاربر را تحلیل می کند، در صورت نیاز ابزار فراخوانی می کند و پاسخ دقیق تولید می کند.

شناخت معماری عامل و نقش LangGraph

عامل هوشمند مجموعه ای از گره ها و وضعیت مشترک است که ورودی کاربر را می گیرد، تصمیم می گیرد به کدام ابزار نیاز دارد و پاسخ می دهد. LangGraph این جریان را با یک گراف قابل برنامه ریزی نشان می دهد تا بتوانید کنترل دقیق بر مسیر گفتگو، استفاده از ابزار و توقف یا ادامه داشته باشید. نتیجه یک ساختار شفاف و تست پذیر است که به راحتی توسعه می یابد.

پیش نیازها و آماده سازی محیط

برای شروع به نسخه های جدید پایتون، یک کلید مدل زبانی و چند بسته نیاز دارید. پیشنهاد می شود یک محیط مجزا بسازید تا وابستگی ها بهم نریزند. سپس کلید مدل را به صورت متغیر محیطی ذخیره کنید تا در کد قرار نگیرد.

1. نصب و ساخت محیط: python -m venv .venv سپس فعال سازی و به روزرسانی pip.
2. نصب کتابخانه ها: langchain، langgraph، langchain openai، python dotenv در صورت نیاز.
3. تنظیم کلید مدل: متغیر OPENAI_API_KEY را در محیط سیستم تنظیم کنید.

# ایجاد محیط مجزا
python -m venv .venv
# فعال سازی در ویندوز
.venv\Scripts\activate
# فعال سازی در مک یا لینوکس
source .venv/bin/activate

# نصب بسته ها
pip install -U langchain langgraph langchain-openai python-dotenv

# تنظیم کلید در شل
# مک و لینوکس:
export OPENAI_API_KEY="your api key"
# ویندوز پاورشل:
setx OPENAI_API_KEY "your api key"

الگوی وضعیت در LangGraph

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

from typing import List, TypedDict
from langchain_core.messages import AnyMessage

class AgentState(TypedDict):
    messages: List[AnyMessage]

ساخت اولین گره عامل و ابزار

در این مرحله یک مدل چت تعریف می کنیم، یک ابزار محاسبات ساده می سازیم و مدل را به ابزارها متصل می کنیم. وقتی مدل تصمیم بگیرد ابزار لازم است، فراخوانی ابزار به صورت ساختاریافته در پیام خروجی درج می شود.

from typing import List
from langchain_core.messages import HumanMessage, AIMessage, AnyMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# مدل چت
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ابزار محاسبه ساده
@tool
def calc(expr: str) -> str:
    """محاسبه ساده عبارت های ریاضی. مثال: 12*7+3"""
    import math
    try:
        allowed = {"sqrt": math.sqrt}
        value = eval(expr, {"__builtins__": {}}, allowed)
        return str(value)
    except Exception as e:
        return f"خطا در محاسبه: {e}"

tools = [calc]
llm_with_tools = llm.bind_tools(tools)

پیاده سازی گراف: گره ها، مسیر شرطی و اجرا

یک گره عامل می سازیم که با مدل تماس می گیرد. سپس یک گره ابزار از پیش ساخته اضافه می کنیم. با یک شرط ساده مشخص می کنیم اگر پیام آخر شامل فراخوانی ابزار بود، به گره ابزار برویم وگرنه پایان یابیم.

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

# تعریف گره عامل
def agent_node(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# گره ابزار آماده
tool_node = ToolNode(tools)

# تابع تصمیم گیری ادامه یا پایان
def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"
    return "end"

# ساخت گراف
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue, {
    "tools": "tools",
    "end": END
})
builder.add_edge("tools", "agent")

app = builder.compile()

اجرای اولین گفتگو

اکنون می توانیم یک پیام انسانی به گراف بدهیم. اگر مدل تشخیص دهد باید محاسبه انجام شود، فراخوانی ابزار ایجاد می کند، گره ابزار آن را اجرا می کند و پاسخ نهایی تولید می شود.

from langchain_core.messages import HumanMessage

# ورودی نمونه
state = {"messages": [HumanMessage(content="حاصل 12*7+3 را حساب کن و در یک جمله توضیح بده")]}

# اجرا و دریافت وضعیت نهایی
final_state = app.invoke(state)

# چاپ پیام ها
for m in final_state["messages"]:
    role = "AI" if isinstance(m, AIMessage) else "User"
    print(role, ":", m.content)

استفاده از جریان رویداد و اشکال زدایی

برای دیدن نحوه حرکت بین گره ها می توانید از استریم رویداد استفاده کنید. این روش برای ردیابی درخواست های ابزار و پاسخ های میانی کاربردی است و اشکال زدایی را ساده می کند.

# مشاهده رویدادها به صورت زنده
for event in app.stream({"messages": [HumanMessage(content="ریشه دوم 144 را حساب کن و نتیجه را توضیح بده")]}):
    for node_name, node_update in event.items():
        print("Node:", node_name, "Update keys:", list(node_update.keys()))

افزودن حافظه گفتگو با Checkpointer

برای مکالمات چند مرحله ای به حافظه نیاز دارید. با Checkpointer می توانید وضعیت را بین نوبت ها نگه دارید. کافی است یک شناسه گفتگو تعیین کنید تا عامل در پیام بعدی سابقه را به خاطر داشته باشد.

from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage

memory = MemorySaver()
app_with_memory = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "session-1"}}

# نوبت اول
app_with_memory.invoke({"messages": [HumanMessage(content="من نامم رضا است")]}, config)

# نوبت دوم با همان شناسه
result = app_with_memory.invoke({"messages": [HumanMessage(content="نام من را به یاد داری؟ اگر بله سلامی دوستانه بده")]}, config)
for m in result["messages"]:
    if isinstance(m, AIMessage):
        print("AI:", m.content)

گسترش توانایی ها با چند ابزار

می توانید ابزارهای بیشتری مانند جستجو در متن محلی، تبدیل تاریخ یا فراخوانی یک API را اضافه کنید. فقط ابزار را با دکوراتور تعریف کنید و در لیست tools قرار دهید. مدل با توجه به دستور کاربر ابزار مناسب را فراخوانی می کند.

from langchain_core.tools import tool

docs = {
    "langgraph": "LangGraph برای ساخت گراف های عامل و مدیریت وضعیت استفاده می شود.",
    "python": "پایتون زبانی ساده و قدرتمند برای ساخت سیستم های داده و هوش مصنوعی است."
}

@tool
def lookup(key: str) -> str:
    """جستجو در دانش محلی بر اساس کلید کوتاه."""
    return docs.get(key.lower(), "چیزی پیدا نشد")

# افزودن ابزار جدید
tools = [calc, lookup]
llm_with_tools = llm.bind_tools(tools)

# بازسازی گره عامل با ابزارهای جدید
def agent_node(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

# ساخت مجدد گراف با ابزار جدید
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
builder.add_edge("tools", "agent")
app = builder.compile()

الگوی کنترل و نگهبان های ساده

گاهی می خواهید پیش از پاسخ نهایی، متن را بررسی یا پاکسازی کنید. می توانید یک گره ساده به عنوان نگهبان بین ابزار و عامل قرار دهید و روی وضعیت پیام آخر منطق اعمال کنید.

def guard_node(state: AgentState) -> AgentState:
    last = state["messages"][-1]
    if isinstance(last, AIMessage) and isinstance(last.content, str):
        content = last.content.strip()
        if len(content) > 1200:
            trimmed = content[:1200] + " ..."
            return {"messages": [AIMessage(content=trimmed, tool_calls=last.tool_calls)]}
    return {}

# افزودن نگهبان به گراف
builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_node("guard", guard_node)
builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": "guard"})
builder.add_edge("tools", "agent")
builder.add_edge("guard", END)
app = builder.compile()

نکات بهینه سازی و الگوهای مفید

برای عملکرد بهتر به چند نکته توجه کنید. اول، ابزارها را حداقل گرا طراحی کنید و ورودی آن ها را اعتبارسنجی کنید. دوم، دما را برای کارهای دقیق روی صفر نگه دارید. سوم، پیام سیستم را برای تعیین نقش عامل شفاف بنویسید.

• مرزبندی وضعیت: فقط کلیدهای ضروری مانند messages را در وضعیت نگه دارید.
• ابزار سبک: پاسخ ابزار باید کوتاه و ساختارمند باشد تا بار توکن کم شود.
• استفاده از شاخه بندی: مسیرهای مجزا برای استنتاج، استفاده از ابزار و جمع بندی نهایی بسازید.
• حافظه انتخابی: فقط رشته گفتگوهای مهم را با checkpointer نگه دارید تا هزینه مدیریت کاهش یابد.
• لاگ گیری رویداد: از استریم رویداد برای بازرسی رفتار عامل در محیط واقعی استفاده کنید.

خطاهای رایج و رفع آن ها

چند خطا بیشتر تکرار می شوند. اگر ابزار کار نمی کند احتمالا مدل به ابزار وصل نشده یا نام پارامترها با شرح ابزار سازگار نیست. اگر وضعیت ادغام نمی شود بررسی کنید گره ها مقدار بازگشتی با کلید messages را ارائه دهند.

• ابزار بدون bind: فراموشی llm.bind_tools باعث می شود مدل هرگز ابزار را صدا نزند.
• ناهماهنگی نوع پیام: استفاده از دیکشنری خام به جای HumanMessage یا AIMessage گاهی مشکل ساز است.
• عدم استفاده از شرط: اگر مسیر شرطی تعریف نشود گراف ممکن است در حلقه بی نهایت بماند.
• نبود thread id: در حافظه چند نوبتی بدون شناسه گفتگو، وضعیت بین کاربران قاطی می شود.

الگوی استقرار سبک

پس از اطمینان از عملکرد، گراف را در یک وب سرویس سبک قرار دهید. یک تابع ساده که پیام کاربر را دریافت می کند، با app.invoke اجرا می کند و پاسخ آخر AI را بر می گرداند کافی است. از پایداری حافظه با شناسه کاربر غافل نشوید.

# اسکچ ساده فست ای پی ای
# pip install fastapi uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class ChatIn(BaseModel):
    user_id: str
    message: str

@app_api.post("/chat")
def chat(inp: ChatIn):
    config = {"configurable": {"thread_id": inp.user_id}}
    out = app.invoke({"messages": [HumanMessage(content=inp.message)]}, config)
    answer = ""
    for m in reversed(out["messages"]):
        if isinstance(m, AIMessage):
            answer = m.content
            break
    return {"reply": answer}

نمونه جریان کامل با دستور کاربر

برای اطمینان از کارکرد صحیح، یک ورودی مرکب امتحان کنید که هم نیاز به ابزار دارد و هم پاسخ نهایی را به صورت متنی می خواهد. این ورودی به مدل کمک می کند همزمان محاسبه و بیان مناسب انجام دهد.

# مثال نهایی
user_prompt = (
    "عدد 19*23 را حساب کن، سپس توضیح کوتاه بده که چطور به نتیجه رسیدی. "
    "اگر لازم است از ابزار استفاده کن."
)
state = {"messages": [HumanMessage(content=user_prompt)]}
final = app.invoke(state)
for m in final["messages"]:
    if isinstance(m, AIMessage):
        print("AI:", m.content)

جمع بندی

در این راهنما یک عامل گفتگو محور با LangGraph ساختیم که با تعریف وضعیت پیام ها، گره عامل، گره ابزار و مسیر شرطی کار می کند. سپس اجرای رویدادی، حافظه چند نوبتی، نگهبان پاسخ و یک اسکچ استقرار را اضافه کردیم تا یک مسیر عملی از ایده تا اجرا شکل بگیرد. با همین الگو می توانید ابزارهای دامنه ای، منطق مسیریابی پیچیده و گزارش گیری را مرحله به مرحله بیفزایید و یک AI Agent با پایتون پایدار و قابل نگهداری بسازید.