Skip to content
> 💻 🧠 קוד 1001 > > מדריך מעשי לבניית סוכני AI עם LangGraph ו-MCP

מדריך מעשי לבניית סוכני AI עם LangGraph ו-MCP

מאמר זה הוא מדריך מעשי למפתחים ליצירת סוכני AI אוטונומיים בפייתון. לא נחזור על התיאוריה של מהם LangChain ו-LangGraph. במקום זאת, נתמקד בקוד, בארכיטקטורה ובפתרון בעיות מהעולם האמיתי.

מטרה: לבנות שני פרויקטים מאפס:

  1. סוכן מסווג (Classifier Agent): סוכן רב-שלבי עם ניהול מצב, אך ללא כלים חיצוניים.
  2. סוכן עוזר (Assistant Agent): סוכן מלא עם גישה למערכת הקבצים ולחיפוש באינטרנט באמצעות פרוטוקול MCP, הבנוי על לוגיקה מחזורית (ציקלית).

נעסוק בשיטות עבודה מומלצות: ניהול תצורה, בחירת מודלים וטיפול בשגיאות ליצירת מערכות חסינות.

In Questo Articolo

בקצרה על המושגים: הסוכן וגשר ה-MCP

לפני שנצלול לקוד, נגדיר שני מושגים:

  • סוכן AI: תוכנית הבנויה סביב לולאת "היגיון-פעולה". היא מקבלת משימה, משתמשת ב-LLM כדי להחליט מה לעשות הלאה (למשל, להפעיל כלי), מבצעת את הפעולה וחוזרת על המחזור עד להשלמת המשימה.
  • MCP (Model Context Protocol): תקן המשמש כגשר בין הלוגיקה של הסוכן לכלים חיצוניים. הוא מאפשר לסוכן לעבוד עם קבצים, ממשקי API או חיפוש באופן אחיד, מבלי לדאוג לפרטי המימוש שלהם.

חלק 1: הקמת סביבה חסינה

שלב 1: סביבה וירטואלית ותלויות

צרו והפעילו סביבה וירטואלית. לאחר מכן, צרו קובץ requirements.txt:

# Frameworks Core
langchain
langgraph

# Model Adapters
langchain-openai
langchain-google-genai
langchain-mistralai
langchain-community # For Ollama

# Tools and Protocols
langchain-mcp-adapters
mcp
ollama

# Utilities
python-dotenv
tenacity # For robust error handling

התקינו את התלויות:

pip install -r requirements.txt```

#### שלב 2: תצורת מפתחות API

צרו קובץ `.env` לאחסון המפתחות שלכם:

OPENAI_API_KEY="sk-…"
GOOGLE_API_KEY="AIzaSy…"
MISTRAL_API_KEY="…"
BRAVE_API_KEY="…" # For the web search tool via MCP

#### שלב 3: תבנית ה-"Model Factory"

כדי לעבור בגמישות בין מודלים בענן למודלים מקומיים מבלי לשנות את קוד הסוכן, נשתמש בתבנית העיצוב "factory".

python

llm_factory.py

import os
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI
from langchain_community.chat_models import ChatOllama

load_dotenv()

class ModelProvider(Enum):
OPENAI = "openai"
GEMINI = "gemini"
MISTRAL_API = "mistral_api"
OLLAMA = "ollama"

def get_llm(provider: ModelProvider, model_name: str = None):
"""A factory for creating LLM instances."""
if provider == ModelProvider.OPENAI:
return ChatOpenAI(model=model_name or "gpt-4o-mini", temperature=0)
elif provider == ModelProvider.GEMINI:
return ChatGoogleGenerativeAI(model=model_name or "gemini-1.5-flash", temperature=0)
elif provider == ModelProvider.MISTRAL_API:
return ChatMistralAI(model=model_name or "mistral-large-latest", temperature=0)
elif provider == ModelProvider.OLLAMA:
# Ensure you have Ollama running with the required model
# docker exec -it ollama ollama pull mistral
return ChatOllama(model=model_name or "mistral", temperature=0)
raise ValueError(f"Unknown model provider: {provider}")

Example usage

if name == "main":
# local_llm = get_llm(ModelProvider.OLLAMA)
openai_llm = get_llm(ModelProvider.OPENAI)
response = openai_llm.invoke("Explain the concept of RAG in three sentences.")
print(response.content)

### חלק 2: פרויקט 1 — סוכן לסיווג משרות

סוכן זה מדגים כיצד להשתמש ב-LangGraph ליצירת **גרף לינארי** עם ניהול מצב. הוא יקבל תיאור משרה ויסווג אותו באופן רציף על פי שלושה פרמטרים.

#### שלב 1: הגדרת המצב (State)

המצב הוא "הזיכרון" של הגרף שלנו, המועבר מצומת לצומת.

python

vacancy_classifier.py

from typing import TypedDict, Dict

class ClassificationState(TypedDict):
"""State for the classifier agent."""
description: str # Source text
job_type: str # Job type (project/permanent)
category: str # Profession
search_type: str # Goal (looking for a job/performer)
classification_log: list # Debug log

#### שלב 2: מימוש צמתי הגרף

כל צומת הוא פונקציה המקבלת את המצב, מבצעת את חלקה בעבודה, ומחזירה את המצב המעודכן.

python
import asyncio
import json
from langchain_core.prompts import ChatPromptTemplate
from llm_factory import get_llm, ModelProvider

class VacancyClassifierAgent:
def init(self):
self.llm = get_llm(ModelProvider.OPENAI, model_name="gpt-4o-mini")

async def _classify_job_type(self, state: ClassificationState) -> ClassificationState:
    """Node 1: Determines the job type."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Determine the job type. The answer must be 'project-based' or 'permanent'."),
        ("human", "Job description:\n\n{description}")
    ])
    chain = prompt | self.llm
    result = await chain.ainvoke({"description": state["description"]})

    state["job_type"] = result.content.strip()
    state["classification_log"].append("Determined job type.")
    return state

async def _classify_category(self, state: ClassificationState) -> ClassificationState:
    """Node 2: Determines the profession category."""
    # Categories can be loaded from a file or database
    categories = ["Python Developer", "Designer", "Marketer", "3D Animator"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"Choose the most suitable category from the list: {', '.join(categories)}."),
        ("human", "Job description:\n\n{description}")
    ])
    chain = prompt | self.llm
    result = await chain.ainvoke({"description": state["description"]})

    state["category"] = result.content.strip()
    state["classification_log"].append("Determined category.")
    return state

async def _classify_search_type(self, state: ClassificationState) -> ClassificationState:
    """Node 3: Determines the search goal."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Determine the author's goal. The answer must be 'looking for a job' or 'looking for a performer'."),
        ("human", "Job description:\n\n{description}")
    ])
    chain = prompt | self.llm
    result = await chain.ainvoke({"description": state["description"]})

    state["search_type"] = result.content.strip()
    state["classification_log"].append("Determined search goal.")
    return state
#### שלב 3: הרכבה והרצת הגרף

אנו מרכיבים את הצמתים לתהליך עבודה יחיד.

python

… continuation of the VacancyClassifierAgent class …

from langgraph.graph import StateGraph, END

def build_graph(self):
    """Assembles the state graph."""
    workflow = StateGraph(ClassificationState)

    workflow.add_node("job_type_classifier", self._classify_job_type)
    workflow.add_node("category_classifier", self._classify_category)
    workflow.add_node("search_type_classifier", self._classify_search_type)

    workflow.set_entry_point("job_type_classifier")
    workflow.add_edge("job_type_classifier", "category_classifier")
    workflow.add_edge("category_classifier", "search_type_classifier")
    workflow.add_edge("search_type_classifier", END)

    return workflow.compile()

async def main():
agent = VacancyClassifierAgent()
graph = agent.build_graph()

description = "We are looking for an experienced Python developer to join our team full-time to work on a fintech project."

initial_state = ClassificationState(
    description=description,
    job_type="", category="", search_type="",
    classification_log=[]
)

final_state = await graph.ainvoke(initial_state)

print("--- Classification Result ---")
print(json.dumps(final_state, indent=2, ensure_ascii=False))

if name == "main":
asyncio.run(main())

### חלק 3: פרויקט 2 — סוכן עוזר עם כלים (MCP)

סוכן זה מדגים **לוגיקה מחזורית**, שבה הוא יכול להפעיל כלים שוב ושוב כדי לפתור משימה.

#### שלב 1: ניהול תצורה

עבור סוכנים המקיימים אינטראקציה עם העולם החיצון, תצורה חסינה היא חיונית.

python

mcp_agent_config.py

from dataclasses import dataclass, field
import os
from llm_factory import ModelProvider

@dataclass
class AgentConfig:
workdir: str = "./agent_workdir"
model_provider: ModelProvider = ModelProvider.OLLAMA

def __post_init__(self):
    """Post-initialization validation."""
    os.makedirs(self.workdir, exist_ok=True)
#### שלב 2: הגדרת המצב לדיאלוג

המצב יאחסן כעת את היסטוריית ההודעות.

python

mcp_agent.py

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]

#### שלב 3: מימוש הגרף המחזורי

הגרף יכלול שני צמתים עיקריים וקשת מותנית היוצרת את לולאת "היגיון-פעולה".

python
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolExecutor
from langchain_mcp_adapters.langchain import V1ToolExecutor
from langchain_mcp_adapters.clients import MultiServerMCPClient
from llm_factory import get_llm
from mcp_agent_config import AgentConfig

class MCPAgent:
def init(self, config: AgentConfig):
self.config = config
self.llm = get_llm(config.model_provider)
self.tools = []
self.tool_executor = None

async def setup_tools(self):
    """Initializes tools via MCP."""
    mcp_config = {
        "filesystem": {
            "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", self.config.workdir],
            "transport": "stdio"
        },
        # Add brave-search if you have a BRAVE_API_KEY
    }
    mcp_client = MultiServerMCPClient(mcp_config)
    self.tools = await mcp_client.get_tools()
    self.tool_executor = ToolExecutor([V1ToolExecutor(tool) for tool in self.tools])

    # Bind the tools to the model
    self.llm = self.llm.bind_tools(self.tools)

def _should_continue(self, state: AgentState):
    """Conditional edge: decides whether to call a tool."""
    last_message = state['messages'][-1]
    if not last_message.tool_calls:
        return "end"
    return "continue"

def _call_model(self, state: AgentState):
    """Node 1: Call the LLM to make a decision."""
    response = self.llm.invoke(state['messages'])
    return {"messages": [response]}

def _call_tool(self, state: AgentState):
    """Node 2: Execute the tool call."""
    last_message = state['messages'][-1]
    tool_call = last_message.tool_calls[0]

    action = {"tool": tool_call["name"], "tool_input": tool_call["args"], "log": ""}
    response = self.tool_executor.invoke(action)

    return {"messages": [response]}

def build_graph(self):
    workflow = StateGraph(AgentState)
    workflow.add_node("agent", self._call_model)
    workflow.add_node("action", self._call_tool)

    workflow.set_entry_point("agent")
    workflow.add_conditional_edges(
        "agent",
        self._should_continue,
        {"continue": "action", "end": END}
    )
    workflow.add_edge("action", "agent")

    return workflow.compile()
#### שלב 4: הרצה ואינטראקציה

python

… continuation of mcp_agent.py …

import asyncio
from langchain_core.messages import HumanMessage
from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
async def run_agent_task(graph, task):
"""Runs a task with error handling."""
return await graph.ainvoke({"messages": [HumanMessage(content=task)]})

async def main():
config = AgentConfig(model_provider=ModelProvider.OPENAI) # or OLLAMA
agent = MCPAgent(config)
await agent.setup_tools()
graph = agent.build_graph()

task = "Create a file named 'hello.txt' in the working directory and write 'Hello, world!' into it."
result = await run_agent_task(graph, task)

print("\n--- Agent's Final Response ---")
print(result['messages'][-1].content)

if name == "main":
asyncio.run(main())
`` כאן, הוספנו את הדקורטורtenacity` לחסינות — אם קריאת הסוכן תיכשל עקב שגיאת רשת זמנית, היא תנסה שוב באופן אוטומטי.

סיכום

בנינו שני סוגים של סוכנים תוך שימוש בשיטות עבודה מודרניות:

  • גרף לינארי מצוין למשימות עם רצף ברור של שלבים, כגון תהליכי ETL או ניתוח רב-שלבי.
  • גרף מחזורי הוא הבסיס ליצירת עוזרים אינטראקטיביים וסוכנים אוטונומיים המסוגלים לפתור בעיות מורכבות בעזרת כלים.

תבניות הארכיטקטורה שהוצגו — מפעל המודלים, ניהול תצורה, הפרדת לוגיקה לצמתים ושימוש בגרפי מצבים — הן היסודות לבניית מערכות AI מדרגיות וחסינות.

Tags:

כתיבת תגובה

האימייל לא יוצג באתר. שדות החובה מסומנים *