پروژه نهایی — ساخت یه Research Agent از صفر

مقدمه: وقتشه یه Agent واقعی و کامل بسازیم

به اپیزود آخر سری خوش اومدی. تا اینجا همه مفاهیم رو یاد گرفتی — از اصول Agent تا ابزار، حافظه، برنامه‌ریزی، Multi-Agent، فریمورک‌ها، امنیت و تست. حالا وقتشه همه اینا رو کنار هم بذاریم و یه پروژه واقعی بسازیم.

می‌خوایم یه Research Agent بسازیم — یه Agent که بتونه وب رو جستجو کنه، اطلاعات جمع‌آوری کنه، خلاصه کنه و یه گزارش کامل بنویسه. از صفر تا آخر.

معماری کلی

Research Agent ما از ۴ بخش اصلی تشکیل شده:

• Planner: وظیفه تحقیق رو به مراحل تقسیم می‌کنه
• Searcher: وب رو جستجو می‌کنه و نتایج رو برمی‌گردونه
• Analyzer: متن‌ها رو می‌خونه و اطلاعات مهم رو استخراج می‌کنه
• Writer: گزارش نهایی رو می‌نویسه

اینا می‌تونن ۴ تا Agent جدا باشن (Multi-Agent) یا ۴ تا مرحله تو یه Agent. ما با رویکرد تک Agent و چند مرحله پیش می‌ریم — ساده‌تره و برای یادگیری بهتره.

قدم ۱: ابزارها

import httpx
import json
from openai import OpenAI
from datetime import datetime

client = OpenAI()

# ابزار ۱: جستجوی وب
async def search_web(query: str, num_results: int = 5) -> str:
    """جستجو تو وب با SerpAPI (یا Tavily)."""
    # اینجا از Tavily API استفاده می‌کنیم
    # ثبت‌نام: tavily.com — پلن رایگان ۱۰۰۰ جستجو در ماه
    
    async with httpx.AsyncClient() as http:
        resp = await http.post(
            "https://api.tavily.com/search",
            json={
                "api_key": TAVILY_API_KEY,
                "query": query,
                "max_results": num_results,
                "include_raw_content": False,
                "search_depth": "advanced",
            },
            timeout=30,
        )
        data = resp.json()
    
    results = []
    for r in data.get("results", []):
        results.append({
            "title": r["title"],
            "url": r["url"],
            "content": r["content"][:500],
        })
    
    return json.dumps(results, ensure_ascii=False, indent=2)


# ابزار ۲: خواندن محتوای صفحه وب
async def read_webpage(url: str) -> str:
    """محتوای یه صفحه وب رو می‌خونه."""
    async with httpx.AsyncClient() as http:
        try:
            resp = await http.get(
                url, 
                follow_redirects=True, 
                timeout=15,
                headers={
                    "User-Agent": "Mozilla/5.0 Research-Agent"
                }
            )
            
            # استخراج متن اصلی (ساده)
            from html.parser import HTMLParser
            
            class TextExtractor(HTMLParser):
                def __init__(self):
                    super().__init__()
                    self.text = []
                    self.skip = False
                    
                def handle_starttag(self, tag, attrs):
                    if tag in ["script", "style", "nav"]:
                        self.skip = True
                        
                def handle_endtag(self, tag):
                    if tag in ["script", "style", "nav"]:
                        self.skip = False
                        
                def handle_data(self, data):
                    if not self.skip:
                        text = data.strip()
                        if text:
                            self.text.append(text)
            
            extractor = TextExtractor()
            extractor.feed(resp.text)
            full_text = "\n".join(extractor.text)
            
            # محدود کردن طول
            return full_text[:3000]
            
        except Exception as e:
            return f"خطا در خواندن صفحه: {str(e)}"


# ابزار ۳: ذخیره یادداشت
notes = []

def save_note(content: str, source: str = "") -> str:
    """یه یادداشت ذخیره می‌کنه."""
    note = {
        "id": len(notes) + 1,
        "content": content,
        "source": source,
        "timestamp": datetime.now().isoformat(),
    }
    notes.append(note)
    return f"یادداشت #{note['id']} ذخیره شد."


def get_all_notes() -> str:
    """همه یادداشت‌ها رو برمی‌گردونه."""
    if not notes:
        return "هنوز هیچ یادداشتی نداری."
    return json.dumps(notes, ensure_ascii=False, indent=2)


# تعریف ابزارها برای OpenAI
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "جستجو در اینترنت. برای پیدا کردن اطلاعات جدید استفاده کن.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "عبارت جستجو (به انگلیسی بهتره)"
                    },
                    "num_results": {
                        "type": "integer",
                        "description": "تعداد نتایج",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_webpage",
            "description": "محتوای یه صفحه وب رو می‌خونه",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "آدرس صفحه وب"
                    }
                },
                "required": ["url"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "save_note",
            "description": "یه یادداشت مهم ذخیره کن. اطلاعات کلیدی رو اینجا بنویس.",
            "parameters": {
                "type": "object",
                "properties": {
                    "content": {
                        "type": "string",
                        "description": "محتوای یادداشت"
                    },
                    "source": {
                        "type": "string",
                        "description": "منبع اطلاعات (URL یا عنوان)"
                    }
                },
                "required": ["content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_all_notes",
            "description": "همه یادداشت‌هایی که تا الان ذخیره کردی رو ببین",
            "parameters": {
                "type": "object",
                "properties": {}
            }
        }
    },
]

# مپ توابع
TOOL_MAP = {
    "search_web": search_web,
    "read_webpage": read_webpage,
    "save_note": save_note,
    "get_all_notes": get_all_notes,
}

قدم ۲: موتور Agent با حلقه ReAct

import asyncio

SYSTEM_PROMPT = """تو یه Research Agent هستی. 
وظیفه‌ات تحقیق درباره موضوعات مختلفه.

روش کارت:
۱. اول یه برنامه تحقیق بنویس (چه سوالاتی باید جواب بدی)
۲. جستجو کن و اطلاعات جمع کن
۳. صفحات مهم رو بخون
۴. نکات کلیدی رو تو یادداشت ذخیره کن
۵. وقتی اطلاعات کافی جمع کردی، گزارش نهایی بنویس

قوانین:
- حداقل ۳ منبع مختلف بررسی کن
- اطلاعات کلیدی رو حتماً تو یادداشت ذخیره کن
- گزارش نهایی باید ساختاریافته باشه (عنوان، مقدمه، بخش‌ها، نتیجه‌گیری)
- گزارش رو به فارسی بنویس
- منابع رو ذکر کن
- وقتی تحقیق تموم شد و گزارش آماده‌ست، پیامت رو با «گزارش نهایی:» شروع کن
"""

MAX_ITERATIONS = 15  # حداکثر تعداد حلقه


async def run_agent(research_topic: str) -> str:
    """Agent رو اجرا می‌کنه تا تحقیق کامل بشه."""
    
    global notes
    notes = []  # ریست یادداشت‌ها
    
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": 
         f"درباره این موضوع تحقیق کن و گزارش بنویس:\n{research_topic}"},
    ]
    
    for iteration in range(MAX_ITERATIONS):
        print(f"\n--- Iteration {iteration + 1} ---")
        
        # فراخوانی LLM
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=TOOLS,
            temperature=0.3,
        )
        
        msg = response.choices[0].message
        
        # اگه ابزار می‌خواد استفاده کنه
        if msg.tool_calls:
            # اضافه کردن پیام assistant
            messages.append({
                "role": "assistant",
                "content": msg.content,
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {
                            "name": tc.function.name,
                            "arguments": tc.function.arguments,
                        }
                    }
                    for tc in msg.tool_calls
                ]
            })
            
            # اجرای همه ابزارها
            for tool_call in msg.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(
                    tool_call.function.arguments
                )
                
                print(f"  Tool: {func_name}({func_args})")
                
                # اجرای ابزار
                func = TOOL_MAP[func_name]
                if asyncio.iscoroutinefunction(func):
                    result = await func(**func_args)
                else:
                    result = func(**func_args)
                
                print(f"  Result: {str(result)[:100]}...")
                
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result),
                })
        else:
            # پاسخ متنی بدون ابزار
            content = msg.content or ""
            messages.append({
                "role": "assistant", 
                "content": content
            })
            
            # چک کن آیا گزارش نهایی آماده‌ست
            if "گزارش نهایی" in content:
                print("\nتحقیق تموم شد!")
                return content
    
    # اگه به حد حلقه رسیدیم
    return (
        "تحقیق به حداکثر تعداد مراحل رسید. "
        "آخرین وضعیت:\n" + messages[-1].get("content", "")
    )

قدم ۳: نوشتن گزارش ساختاریافته

async def generate_report(
    topic: str, notes: list
) -> str:
    """از یادداشت‌ها یه گزارش حرفه‌ای تولید می‌کنه."""
    
    notes_text = "\n".join(
        f"- [{n['source']}] {n['content']}" 
        for n in notes
    )
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """
            تو یه نویسنده گزارش حرفه‌ای هستی.
            از یادداشت‌های تحقیقاتی، یه گزارش کامل 
            و ساختاریافته بنویس.
            
            ساختار گزارش:
            ۱. عنوان
            ۲. خلاصه اجرایی (۳-۴ جمله)
            ۳. مقدمه
            ۴. یافته‌های اصلی (چند بخش)
            ۵. تحلیل و بررسی
            ۶. نتیجه‌گیری
            ۷. منابع
            
            به فارسی بنویس. لحن حرفه‌ای ولی خوانا.
            """},
            {"role": "user", "content": 
             f"موضوع: {topic}\n\nیادداشت‌ها:\n{notes_text}"},
        ],
        temperature=0.4,
        max_tokens=3000,
    )
    
    return response.choices[0].message.content

قدم ۴: اجرای کامل

async def research(topic: str):
    """فرآیند کامل تحقیق."""
    
    print(f"شروع تحقیق: {topic}")
    print("=" * 60)
    
    # اجرای Agent
    report = await run_agent(topic)
    
    # ذخیره گزارش
    filename = f"report_{datetime.now():%Y%m%d_%H%M}.md"
    with open(filename, "w") as f:
        f.write(report)
    
    print(f"\nگزارش ذخیره شد: {filename}")
    print(f"تعداد یادداشت‌ها: {len(notes)}")
    
    return report

# اجرا
# asyncio.run(research(
#     "تأثیر هوش مصنوعی بر بازار کار ایران در سال ۲۰۲۵"
# ))

قدم ۵: اضافه کردن Guardrails

class ResearchGuardrails:
    """محافظ‌های مخصوص Research Agent."""
    
    MAX_SEARCHES = 10
    MAX_PAGE_READS = 8
    MAX_COST = 0.50  # دلار
    BLOCKED_DOMAINS = [
        "facebook.com", "instagram.com", 
        "tiktok.com",  # شبکه‌های اجتماعی منبع معتبر نیستن
    ]
    
    def __init__(self):
        self.search_count = 0
        self.read_count = 0
        self.estimated_cost = 0.0
    
    def check_search(self, query: str) -> tuple[bool, str]:
        if self.search_count >= self.MAX_SEARCHES:
            return False, "به حداکثر جستجو رسیدی."
        self.search_count += 1
        return True, ""
    
    def check_read(self, url: str) -> tuple[bool, str]:
        if self.read_count >= self.MAX_PAGE_READS:
            return False, "به حداکثر خواندن صفحه رسیدی."
        
        for domain in self.BLOCKED_DOMAINS:
            if domain in url:
                return False, f"دامنه {domain} مجاز نیست."
        
        self.read_count += 1
        return True, ""
    
    def check_cost(self, tokens: int) -> tuple[bool, str]:
        # تخمین هزینه gpt-4o
        cost = tokens * 0.005 / 1000
        self.estimated_cost += cost
        
        if self.estimated_cost >= self.MAX_COST:
            return False, "به حداکثر بودجه رسیدی."
        return True, ""

بهبودهای پیشرفته

این نسخه پایه‌ست. چند ایده برای بهترش کردن:

۱. حافظه بلندمدت: نتایج تحقیقات قبلی رو تو Vector Database ذخیره کن. دفعه بعد که موضوع مشابهی بخوای تحقیق کنی، از نتایج قبلی هم استفاده کنه.

۲. Multi-Agent: به جای یه Agent، ۳ تا Agent جدا بساز — Searcher، Analyzer، Writer — و یه Orchestrator بذار هماهنگشون کنه.

۳. منابع بیشتر: علاوه بر وب، از API های خبری، ویکی‌پدیا، و دیتابیس‌های علمی (مثل arXiv) هم جستجو کنه.

۴. فرمت خروجی: گزارش رو به فرمت PDF، HTML یا حتی اسلاید تبدیل کن.

۵. ارزیابی خودکار: بعد از نوشتن گزارش، یه Agent دیگه کیفیتش رو بررسی کنه و اگه ضعیف بود، دوباره بنویسه.

جمع‌بندی سری — از اپیزود ۱ تا ۱۰

بذار یه خلاصه از کل سری بدم:

اپیزود ۱: Agent چیه و چه فرقی با Chatbot داره — فهمیدیم Agent تصمیم می‌گیره و عمل می‌کنه.

اپیزود ۲: Tool Use — بهش دست و پا دادیم.

اپیزود ۳: RAG و حافظه — بهش مغز دادیم.

اپیزود ۴: Planning — بهش قدرت فکر کردن دادیم.

اپیزود ۵: Multi-Agent — یه تیم از Agent ها ساختیم.

اپیزود ۶: فریمورک‌ها — ابزارهای آماده رو شناختیم.

اپیزود ۷: امنیت — Agent رو امن کردیم.

اپیزود ۸: بات تلگرامی — یه Agent واقعی ساختیم.

اپیزود ۹: تست و دیباگ — یاد گرفتیم مطمئن بشیم کار می‌کنه.

اپیزود ۱۰ (همین): Research Agent — یه پروژه کامل از صفر.

ممنون که تا اینجا همراه من بودی. امیدوارم این سری بهت کمک کرده باشه مفاهیم Agent رو عمیق بفهمی و عملی پیاده‌سازی کنی. اگه سوالی داری، زیر همین اپیزود بنویس.