بناء أنظمة وكلاء متعددة جاهزة للإنتاج باستخدام Agno في بايثون

بناء وكيل ذكاء اصطناعي واحد أمر سهل. أما بناء نظام من الوكلاء يخطط ويفوّض المهام ويتذكر ويعمل بموثوقية في بيئة الإنتاج، فهنا تتعثر معظم المشاريع. وقد بُني إطار Agno — المعروف سابقًا باسم Phidata — لسدّ هذه الفجوة تحديدًا. يُنشأ الوكلاء في أجزاء من المليون من الثانية، ويستهلكون جزءًا يسيرًا من ذاكرة أطر العمل الأثقل، ويأتون مزوّدين بكل ما تحتاجه: الذاكرة، وقواعد المعرفة، والمخرجات المهيكلة، وزمن تشغيل إنتاجي اسمه AgentOS يغلّف كل شيء خلف خادم FastAPI.

في هذا الدرس ستبني مساعد بحث ينمو من وكيل واحد إلى فريق منسّق، ثم إلى سير عمل متعدد الخطوات، وأخيرًا إلى واجهة برمجية قابلة للنشر. وبنهاية الدرس ستفهم العناصر الأربعة الأساسية في Agno — الوكيل (Agent)، والفريق (Team)، وسير العمل (Workflow)، وAgentOS — وكيف تتركّب معًا.

المتطلبات المسبقة

قبل البدء، تأكد من توفّر ما يلي:

• بايثون 3.10 أو أحدث (python --version)
• إلمام أساسي ببايثون وبصيغة async/await
• مفتاح OpenAI API (أو Anthropic — فإطار Agno محايد تجاه النماذج)
• طرفية ومحرر أكواد (يُفضَّل VS Code)

يُستحسن أن تكون مرتاحًا في التعامل مع البيئات الافتراضية وقراءة نماذج Pydantic. لا حاجة لخبرة سابقة بأطر عمل الوكلاء.

ما الذي ستبنيه

مساعد بحث يزداد ثراءً تدريجيًا:

1. وكيل واحد يبحث في الويب ويجيب عن الأسئلة مع ذكر المصادر.
2. فريق يفوّض فيه القائد المهام إلى وكلاء متخصصين (بحث في الويب + تحليل).
3. سير عمل يشغّل البحث والتحليل بالتوازي، ويكرّر حتى يجمع مادة كافية، ثم يكتب تقريرًا.
4. نشر عبر AgentOS يكشف كل ما سبق كواجهة REST مع جلسات دائمة.

الخطوة 1: تجهيز المشروع

أنشئ مجلد المشروع وبيئة افتراضية:

mkdir agno-research && cd agno-research
python -m venv .venv
source .venv/bin/activate   # على ويندوز: .venv\Scripts\activate

ثبّت Agno مع مزوّد نموذج والأدوات التي سنستخدمها:

pip install -U agno openai ddgs lancedb tantivy

ملاحظة سريعة عمّا تقوم به كل حزمة:

• agno — إطار العمل نفسه
• openai — عميل مزوّد النموذج
• ddgs — يشغّل WebSearchTools (بحث DuckDuckGo، دون الحاجة لمفتاح)
• lancedb وtantivy — قاعدة البيانات المتجهة المدمجة لقواعد المعرفة (الخطوة 5)

صدّر مفتاح الـ API:

export OPENAI_API_KEY="sk-..."

نصيحة: إطار Agno محايد فعلًا تجاه النماذج. لاستخدام Claude بدلًا من ذلك، نفّذ pip install anthropic، واضبط ANTHROPIC_API_KEY، واستبدل OpenAIChat(id="gpt-4o-mini") بـ Claude(id="claude-sonnet-4-0"). كل ما تبقّى في هذا الدرس يبقى كما هو.

الخطوة 2: وكيلك الأول

الـ Agent في Agno هو الوحدة الذرّية: نموذج، ومجموعة تعليمات، وأدوات اختيارية. أنشئ ملف agent.py:

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.websearch import WebSearchTools
 
web_agent = Agent(
    name="Web Researcher",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[WebSearchTools()],
    instructions=[
        "You are a meticulous research assistant.",
        "Always search the web before answering factual questions.",
        "Cite your sources as a bulleted list at the end.",
    ],
    add_datetime_to_context=True,
    markdown=True,
)
 
web_agent.print_response(
    "What were the biggest open-source AI agent frameworks released in 2026?",
    stream=True,
)

شغّله:

python agent.py

سترى الوكيل يستدلّ، ويستدعي أداة البحث، ويبثّ إجابة منسّقة مع مصادرها إلى طرفيتك. بعض النقاط الجديرة بالملاحظة:

• instructions قائمة من التوجيهات القصيرة. يجمّعها Agno في موجّه النظام — فاجعلها أمرية ومحدّدة.
• tools مجرّد قائمة من كائنات الأدوات. النموذج هو من يقرّر متى يستدعيها؛ لن تكتب منطق التنسيق يدويًا.
• add_datetime_to_context=True يحقن التاريخ الحالي، مما يُبقي استعلامات "الأحدث" مرتكزة على الواقع.
• print_response(..., stream=True) أسرع وسيلة لرؤية المخرجات أثناء التطوير.

تشغيل الوكلاء برمجيًا

print_response موجّهة للبشر. في الكود الفعلي استخدم run() للحصول على كائن استجابة:

response = web_agent.run("Summarize the Agno framework in two sentences.")
print(response.content)        # الإجابة النصية

ولخدمات عالية الإنتاجية، لكل دالة توأم غير متزامن — arun() وaprint_response() — كي تنتظر تشغيل عدة وكلاء بالتوازي.

الخطوة 3: المخرجات المهيكلة باستخدام Pydantic

النص الناتج عن النماذج اللغوية يصعب استهلاكه برمجيًا. يستطيع Agno إجبار الوكيل على إرجاع كائن مُحدَّد النوع بتمرير نموذج Pydantic عبر output_schema. أنشئ ملف structured.py:

from typing import List
from pydantic import BaseModel, Field
 
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.websearch import WebSearchTools
 
 
class TopicBriefing(BaseModel):
    topic: str = Field(..., description="The subject being researched")
    summary: str = Field(..., description="A two-sentence overview")
    key_points: List[str] = Field(..., description="3-5 essential takeaways")
    sources: List[str] = Field(..., description="URLs used for the research")
 
 
briefing_agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[WebSearchTools()],
    output_schema=TopicBriefing,
)
 
response = briefing_agent.run("Research the rise of small language models in 2026")
briefing: TopicBriefing = response.content
 
print(briefing.topic)
for point in briefing.key_points:
    print(f"- {point}")
print("Sources:", ", ".join(briefing.sources))

ولأن response.content صار الآن نسخة حقيقية من TopicBriefing، تحصل على الإكمال التلقائي والتحقّق وانعدام تحليل النصوص الهشّ. هذه الميزة وحدها تجعل وكلاء Agno آمنين للدمج داخل تطبيقات أكبر.

الخطوة 4: تنسيق فريق

الوكيل الواحد محدود. يتيح لك الفريق (Team) أن يفوّض نموذج قائد المهام إلى أعضاء متخصصين، لكلٍّ منهم دوره وأدواته. يقرّر القائد من يتولّى ماذا، ثم يصوغ الإجابة النهائية.

أنشئ ملف team.py:

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.team import Team
from agno.tools.websearch import WebSearchTools
from agno.tools.hackernews import HackerNewsTools
 
web_agent = Agent(
    name="Web Agent",
    role="Search the broad web for background and context",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[WebSearchTools()],
    instructions="Always include sources.",
)
 
tech_pulse_agent = Agent(
    name="Tech Pulse Agent",
    role="Gauge developer sentiment from Hacker News discussions",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[HackerNewsTools()],
    instructions="Summarize what practitioners actually think, not just headlines.",
)
 
research_team = Team(
    name="Research Team",
    model=OpenAIChat(id="gpt-4o"),
    members=[web_agent, tech_pulse_agent],
    instructions=[
        "You coordinate a research team.",
        "Delegate broad context to the Web Agent.",
        "Delegate community sentiment to the Tech Pulse Agent.",
        "Combine both into a balanced briefing with a clear verdict.",
    ],
    show_members_responses=True,
    markdown=True,
)
 
research_team.print_response(
    "Should a startup adopt Agno for its agent stack in 2026?",
    stream=True,
)

الأفكار الأساسية:

• لكل عضو role — وصف من سطر واحد يستخدمه القائد لتوجيه العمل. اكتب الأدوار كوصف وظيفي، لا كتعليمات.
• يحصل القائد (research_team) عادةً على نموذج أقوى من الأعضاء، لأن التنسيق والصياغة أصعب من عمليات البحث الفردية.
• show_members_responses=True يُظهر مساهمة كل عضو، وهو أمر بالغ القيمة أثناء تصحيح التفويض.

شغّله وراقب القائد وهو يوزّع المهمة على المتخصصين، ثم يدمج نتائجهما في حكم واحد. لم تكتب أي كود توجيه — يتولّى قائد الفريق التفويض عبر استدعاءات الأدوات في الخلفية.

الخطوة 5: إضافة الذاكرة والمعرفة

المساعدون الحقيقيون يتذكرون المحادثات السابقة ويمكنهم تأسيس الإجابات على مستنداتك الخاصة. يتعامل Agno مع الأمرين عبر قاعدة بيانات (للذاكرة والجلسات) وقاعدة معرفة (Knowledge) للاسترجاع.

أنشئ ملف memory_knowledge.py:

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb
from agno.knowledge.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.vectordb.lancedb import LanceDb, SearchType
 
# مخزن دائم للجلسات وذاكرة المستخدم طويلة الأمد
db = SqliteDb(db_file="tmp/research.db")
 
# قاعدة معرفة مدعومة بمخزن متجهات LanceDB مدمج
knowledge = Knowledge(
    vector_db=LanceDb(
        uri="tmp/lancedb",
        table_name="research_docs",
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)
 
# استوعب مستندًا مرة واحدة؛ يُقطَّع ويُضمَّن ويُفهرس تلقائيًا
knowledge.insert(url="https://www.paulgraham.com/read.html")
 
assistant = Agent(
    name="Grounded Assistant",
    model=OpenAIChat(id="gpt-4o-mini"),
    db=db,
    knowledge=knowledge,
    search_knowledge=True,          # السماح للوكيل بالاسترجاع من قاعدة المعرفة
    add_history_to_context=True,    # تضمين الأدوار الأخيرة في الموجّه
    num_history_runs=3,             # كم محادثة سابقة نُضمّن
    update_memory_on_run=True,      # تعلّم حقائق دائمة عن المستخدم
    markdown=True,
)
 
# مرّر user_id و session_id ثابتين لربط الذاكرة عبر عمليات التشغيل
assistant.print_response(
    "According to the essay I added, why does reading matter?",
    user_id="anis@example.com",
    session_id="session-1",
)

ماذا يفعل كل جزء:

• SqliteDb يحفظ الجلسات والذاكرة في ملف محلي. في الإنتاج تستبدله بـ PostgresDb — وكود الوكيل لا يتغير.
• Knowledge + LanceDb يمنحك بحثًا هجينًا (متجهات + كلمات مفتاحية) على المحتوى المُستوعَب. وsearch_knowledge=True تخبر الوكيل أنه يجوز له الاسترجاع منها.
• add_history_to_context و**num_history_runs** تتحكمان في الذاكرة الحوارية داخل الجلسة.
• update_memory_on_run تتيح للوكيل استخلاص حقائق دائمة ("المستخدم يفضّل الإجابات المختصرة") واستدعاءها في جلسات لاحقة مرتبطة بنفس الـ user_id.

هذه هي اللحظة التي يتحوّل فيها العرض التوضيحي إلى مساعد فعلي: يتذكر مَن يحدّثه ويستطيع الاستشهاد بمستنداتك.

الخطوة 6: تنسيق سير عمل

الفرق ممتازة حين تريد أن يقرّر النموذج من يفعل ماذا. لكن أحيانًا تحتاج تدفّق تحكّم حتميًا — شغّل هذه الخطوات بالتوازي، كرّر حتى بلوغ معيار جودة، تفرّع بناءً على شرط. لذلك وُجِد سير العمل (Workflow).

أنشئ ملف workflow.py:

from typing import List
 
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.tools.websearch import WebSearchTools
from agno.tools.hackernews import HackerNewsTools
from agno.workflow import Loop, Parallel, Step, Workflow
from agno.workflow.types import StepOutput
 
# --- الوكلاء ---
web_researcher = Agent(
    name="Web Researcher",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[WebSearchTools()],
    instructions="Research the topic thoroughly from web sources.",
)
hn_researcher = Agent(
    name="HN Researcher",
    model=OpenAIChat(id="gpt-4o-mini"),
    tools=[HackerNewsTools()],
    instructions="Surface what developers are saying about the topic.",
)
writer = Agent(
    name="Report Writer",
    model=OpenAIChat(id="gpt-4o"),
    instructions="Write a concise, well-structured report from the research provided.",
    markdown=True,
)
 
# --- الخطوات ---
research_web = Step(name="Research Web", agent=web_researcher)
research_hn = Step(name="Research HN", agent=hn_researcher)
write_report = Step(name="Write Report", agent=writer)
 
 
# --- شرط الخروج من الحلقة: واصل البحث حتى تجمع مادة كافية ---
def enough_research(outputs: List[StepOutput]) -> bool:
    total = sum(len(o.content or "") for o in outputs)
    return total > 1500
 
 
# --- سير العمل: بحث متوازٍ داخل حلقة، ثم الكتابة ---
workflow = Workflow(
    name="Deep Research Workflow",
    description="Research a topic in parallel until sufficient, then write a report",
    steps=[
        Loop(
            name="Research Loop",
            steps=[Parallel(research_web, research_hn, name="Parallel Research")],
            end_condition=enough_research,
            max_iterations=3,
        ),
        write_report,
    ],
)
 
if __name__ == "__main__":
    workflow.print_response(
        input="The state of TypeScript-first AI agent frameworks in 2026",
        stream=True,
    )

العناصر التي يمنحك إياها Agno لسير العمل:

هنا يعمل الباحثان في الوقت نفسه داخل Loop يستمر حتى يتجاوز مجموع المخرجات 1500 حرف (أو تمرّ ثلاث تكرارات)، وبعدها يُنتج الكاتب التقرير النهائي. وعلى خلاف الفريق، يكون تدفّق التحكّم حتميًا وقابلًا للاختبار بالكامل — إذ تستطيع اختبار دالة enough_research() بمعزل.

التفرّع باستخدام Condition

للتوجيه، يدعم Agno حتى الشروط المبنية على التعبيرات:

from agno.workflow import Condition, Step, Workflow
 
workflow = Workflow(
    name="Classify and Route",
    steps=[
        Step(name="Classify", agent=classifier),
        Condition(
            name="Route by Type",
            evaluator='previous_step_content.contains("TECHNICAL")',
            steps=[Step(name="Technical Help", agent=technical_agent)],
            else_steps=[Step(name="General Help", agent=general_agent)],
        ),
    ],
)

يصنّف المصنّف الطلب، فيرسله Condition إلى الفرع الصحيح — دون أي كود وصل بينهما.

الخطوة 7: النشر باستخدام AgentOS

السكربت العامل ليس منتجًا. يغلّف AgentOS وكلاءك وفِرقك وسير عملك داخل تطبيق FastAPI جاهز — مكتمل بتخزين الجلسات وسجل المحادثات ونقاط نهاية للمراقبة — تشغّله في بنيتك التحتية الخاصة.

أنشئ ملف serve.py:

from agno.agent import Agent
from agno.team import Team
from agno.workflow import Step, Workflow
from agno.models.openai import OpenAIChat
from agno.db.sqlite import SqliteDb
from agno.tools.websearch import WebSearchTools
from agno.os import AgentOS
 
db = SqliteDb(db_file="tmp/agentos.db")
 
researcher = Agent(
    name="Researcher",
    model=OpenAIChat(id="gpt-4o-mini"),
    db=db,
    tools=[WebSearchTools()],
    instructions="Research thoroughly and cite sources.",
    add_history_to_context=True,
    markdown=True,
)
 
research_team = Team(
    name="Research Team",
    model=OpenAIChat(id="gpt-4o"),
    db=db,
    members=[researcher],
    instructions="Coordinate research and deliver a clear briefing.",
)
 
qa_workflow = Workflow(
    name="QA Workflow",
    description="Answer a question using the researcher agent",
    db=db,
    steps=[Step(name="Answer", agent=researcher)],
)
 
agent_os = AgentOS(
    description="Research AgentOS",
    agents=[researcher],
    teams=[research_team],
    workflows=[qa_workflow],
)
 
# نسخة تطبيق FastAPI — يبحث uvicorn عن `app`
app = agent_os.get_app()
 
if __name__ == "__main__":
    agent_os.serve(app="serve:app", reload=True)

شغّل الخادم:

python serve.py

يُقلع AgentOS خادم FastAPI (الافتراضي http://localhost:7777) بنقاط نهاية REST مولّدة تلقائيًا لكل وكيل وفريق وسير عمل سجّلته، إضافةً إلى توثيق تفاعلي على /docs. ولأن كل مكوّن يتشارك نفس db، تُحفظ الجلسات والذاكرة تلقائيًا — فيتذكّر وكلاؤك المحادثات عبر طلبات HTTP.

زمن التشغيل عديم الحالة وقابل للتوسّع الأفقي: شغّل عدة نسخ خلف موازن أحمال ووجّهها إلى قاعدة بيانات Postgres مشتركة. هذا هو "الجزء المفقود" الذي بُني Agno حوله — الجسر من نموذج أوّلي عامل إلى منتج منشور.

اختبار التنفيذ

تحقّق من كل طبقة بالترتيب:

1. الوكيل — python agent.py يبثّ إجابة مع مصادر.
2. المخرجات المهيكلة — python structured.py يطبع حقولًا مُحدَّدة النوع دون أخطاء.
3. الفريق — python team.py يُظهر مساهمة العضوين، ثم حكمًا مدموجًا.
4. الذاكرة — شغّل memory_knowledge.py مرتين بنفس الـ user_id؛ ينبغي أن يستدعي التشغيل الثاني السياق.
5. سير العمل — python workflow.py يشغّل بحثًا متوازيًا داخل حلقة قبل الكتابة.
6. AgentOS — افتح http://localhost:7777/docs واستدعِ نقطة نهاية؛ تأكّد من الاستجابة ومن ظهور صف جلسة في tmp/agentos.db.

استكشاف الأخطاء وإصلاحها

ModuleNotFoundError: No module named 'ddgs' — تحتاج WebSearchTools إلى محرّك البحث. نفّذ pip install ddgs.

استدعاءات أدوات فارغة أو مرفوضة — قد يكون نموذجك أصغر من أن يستدلّ على الأدوات بموثوقية. انقل القائد/المنسّق إلى نموذج أقوى (gpt-4o، claude-sonnet-4-0) وأبقِ النموذج الرخيص لأدوار الأعضاء البسيطة فقط.

قاعدة المعرفة لا تُرجع شيئًا — تأكّد من search_knowledge=True على الوكيل ومن نجاح knowledge.insert(...) قبل الاستعلام. كما يحتاج البحث الهجين إلى تثبيت tantivy.

الذاكرة لا تُحفظ — تأكّد من تمرير user_id وsession_id ثابتين، ومن امتلاك الوكيل لـ db، ومن ضبط update_memory_on_run=True و/أو add_history_to_context=True.

AgentOS لا يُقلع مع serve() — يجب أن تكون سلسلة app= بصيغة "<module>:app" مطابقة لاسم ملفك. في serve.py تكون "serve:app".

الخطوات التالية

• استبدل SqliteDb بـ PostgresDb وانشر AgentOS خلف موازن أحمال.
• أضف خطوات Router لاختيار استراتيجيات البحث ديناميكيًا حسب الموضوع.
• اربط أدوات خارجية عبر MCP باستخدام صنف MCPTools في Agno.
• أضف التقييم عبر ReliabilityEval في Agno لرصد التراجعات في سلوك الوكلاء.
• استكشف دروسًا ذات صلة على موقعنا: أنظمة CrewAI متعددة الوكلاء، ووكلاء Pydantic AI الآمنون نوعيًا، ووكلاء smolagents المرتكزون على الكود.

الخلاصة

لقد نقلت مساعد بحث من وكيل واحد يستخدم أداةً إلى واجهة برمجية قابلة للتوسّع الأفقي. وعلى طول الطريق تعرّفت على عناصر Agno الأربعة التي تتركّب لتكوّن أي نظام وكلائي تقريبًا: الوكيل للقدرة الفردية، والفريق للتفويض المدفوع بالنموذج، وسير العمل للتنسيق الحتمي، وAgentOS للنشر الإنتاجي. ونقاط قوة Agno المميزة — الإنشاء في أجزاء من المليون من الثانية، وبصمة الذاكرة المنخفضة، وزمن التشغيل المزوّد بكل ما يلزم — تعني أن النظام الذي بنيته أوّليًا بعد ظهر اليوم هو نفسه الذي ستنشره. ابدأ صغيرًا بوكيل واحد، ولا تلجأ إلى الفرق وسير العمل إلا حين يتطلّب ذلك المشكلةُ فعلًا.