المنطق الأساسي

الآن لننفذ عقل الوكيل—حلقة ReAct التي تستدل، وتتصرف، وتجمع نتائج البحث.

قوالب المطالبات

# prompts/templates.py

SYSTEM_PROMPT = """أنت مساعد بحث يساعد المستخدمين على فهم المواضيع بشكل شامل.

لديك وصول لهذه الأدوات:
{tools}

عند البحث، اتبع هذه الخطوات:
1. قسّم الموضوع إلى أسئلة رئيسية
2. ابحث عن معلومات للإجابة على كل سؤال
3. اجمع النتائج في سرد متماسك
4. استشهد دائماً بمصادرك

استخدم هذا التنسيق:
فكرة: [استدلالك حول ما يجب فعله بعد ذلك]
إجراء: [اسم_الأداة]
مدخل الإجراء: [استعلام للأداة]

بعد جمع معلومات كافية، قدم إجابتك النهائية بـ:
الإجابة النهائية: [ردك الشامل مع الاستشهادات]
"""

SYNTHESIS_PROMPT = """بناءً على نتائج البحث التالية، اكتب تقريراً شاملاً عن "{topic}".

النتائج:
{findings}

المتطلبات:
- ابدأ بنظرة عامة موجزة
- نظّم في أقسام منطقية
- ضمّن حقائق وبيانات محددة
- استشهد بالمصادر باستخدام [1]، [2]، إلخ.
- اختم بالنقاط الرئيسية
- احتفظ بأقل من {max_length} كلمة

المصادر:
{sources}
"""

مخزن الذاكرة

# memory/store.py
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

@dataclass
class ResearchFinding:
    query: str
    content: str
    source_url: str
    source_title: str
    timestamp: datetime

class ResearchMemory:
    def __init__(self):
        self.findings: List[ResearchFinding] = []
        self.queries_made: set = set()

    def add_finding(self, query: str, content: str, url: str, title: str):
        finding = ResearchFinding(
            query=query,
            content=content,
            source_url=url,
            source_title=title,
            timestamp=datetime.now()
        )
        self.findings.append(finding)
        self.queries_made.add(query)

    def get_all_findings(self) -> str:
        return "\n\n".join([
            f"الاستعلام: {f.query}\n"
            f"المصدر: {f.source_title}\n"
            f"المحتوى: {f.content}"
            for f in self.findings
        ])

    def get_sources(self) -> List[dict]:
        seen = set()
        sources = []
        for i, f in enumerate(self.findings, 1):
            if f.source_url not in seen:
                sources.append({
                    "id": i,
                    "title": f.source_title,
                    "url": f.source_url
                })
                seen.add(f.source_url)
        return sources

    def has_searched(self, query: str) -> bool:
        return query.lower() in {q.lower() for q in self.queries_made}

فئة الوكيل الرئيسية

# agent.py
import re
from langchain_openai import ChatOpenAI
from tools.search import WebSearchTool
from memory.store import ResearchMemory
from prompts.templates import SYSTEM_PROMPT, SYNTHESIS_PROMPT
from config import Config

class ResearchAgent:
    def __init__(self, config: Config):
        self.config = config
        self.llm = ChatOpenAI(
            model=config.MODEL_NAME,
            temperature=config.TEMPERATURE,
            api_key=config.OPENAI_API_KEY
        )
        self.tools = {
            "web_search": WebSearchTool()
        }
        self.memory = None

    def research(self, topic: str) -> str:
        """نقطة الدخول الرئيسية للبحث"""
        self.memory = ResearchMemory()

        # تشغيل حلقة ReAct
        self._research_loop(topic)

        # تجميع النتائج
        report = self._synthesize_report(topic)

        return report

    def _research_loop(self, topic: str):
        """حلقة ReAct لجمع المعلومات"""
        tools_desc = "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in self.tools.items()
        ])

        messages = [
            {"role": "system", "content": SYSTEM_PROMPT.format(tools=tools_desc)},
            {"role": "user", "content": f"ابحث في هذا الموضوع بشكل شامل: {topic}"}
        ]

        for iteration in range(self.config.MAX_ITERATIONS):
            response = self.llm.invoke(messages)
            content = response.content

            # التحقق من الإجابة النهائية
            if "الإجابة النهائية:" in content or "Final Answer:" in content:
                break

            # تحليل الإجراء
            action_match = re.search(
                r"Action:\s*(\w+)\s*\nAction Input:\s*(.+?)(?=\n|$)",
                content,
                re.DOTALL
            )

            if action_match:
                tool_name = action_match.group(1).strip()
                tool_input = action_match.group(2).strip()

                # تنفيذ الأداة
                if tool_name in self.tools:
                    result = self.tools[tool_name].run(tool_input)

                    # تخزين النتائج في الذاكرة
                    if result["success"]:
                        for r in result["results"]:
                            self.memory.add_finding(
                                query=tool_input,
                                content=r["snippet"],
                                url=r["url"],
                                title=r["title"]
                            )

                    # إضافة الملاحظة للرسائل
                    observation = f"ملاحظة: {self._format_results(result)}"
                    messages.append({"role": "assistant", "content": content})
                    messages.append({"role": "user", "content": observation})
                else:
                    messages.append({
                        "role": "user",
                        "content": f"خطأ: أداة غير معروفة '{tool_name}'"
                    })

    def _format_results(self, result: dict) -> str:
        if not result["success"]:
            return f"فشل البحث: {result.get('error', 'خطأ غير معروف')}"

        if not result["results"]:
            return "لم يتم العثور على نتائج"

        formatted = []
        for r in result["results"]:
            formatted.append(f"- {r['title']}: {r['snippet'][:200]}...")

        return "\n".join(formatted)

    def _synthesize_report(self, topic: str) -> str:
        """توليد التقرير النهائي من النتائج"""
        findings = self.memory.get_all_findings()
        sources = self.memory.get_sources()

        sources_text = "\n".join([
            f"[{s['id']}] {s['title']}: {s['url']}"
            for s in sources
        ])

        prompt = SYNTHESIS_PROMPT.format(
            topic=topic,
            findings=findings,
            sources=sources_text,
            max_length=self.config.REPORT_MAX_LENGTH
        )

        response = self.llm.invoke([{"role": "user", "content": prompt}])

        return response.content

قرارات التصميم الرئيسية

القرار	المبرر
نمط ReAct	استدلال شفاف، قابل للتحكم
ذاكرة منفصلة	نتائج مستمرة عبر التكرارات
تجريد الأدوات	سهولة إضافة مصادر بحث جديدة
خطوة التجميع	جودة أفضل من الإخراج المتدفق

التالي: إضافة الاختبار والتحقق لضمان نتائج موثوقة. :::