بنية العميل الذكي (AI Agent): العملاء المستقلون وإطار عمل LangChain

العملاء الذكيون (AI Agents) هم أنظمة ذكاء اصطناعي قادرة على اتخاذ القرارات بشكل مستقل، واستخدام الأدوات، وحل المهام المعقدة خطوة بخطوة. في هذا الدليل، نستعرض بنية العملاء وكيفية تنفيذهم.

ما هو العميل (Agent)؟

مقارنة بسيطة بين روبوت الدردشة والعميل:

1Chatbot:
2User: "How is the weather today?"
3Bot: "I don't know the weather." (static response)
4
5Agent:
6User: "How is the weather today?"
7Agent: 
8  1. "I need to find out the weather."
9  2. Call weather_api.get_weather()
10  3. Interpret the result
11  4. "It's 15°C in Istanbul, partly cloudy."

مكونات العميل

1. LLM (الدماغ)

اتخاذ القرار والاستدلال.

2. الأدوات

التفاعل مع العالم الخارجي:

• البحث على الويب
• استدعاءات API
• استعلامات قواعد البيانات
• تنفيذ الأكواد

3. الذاكرة

سجل المحادثات وتخزين المعلومات.

4. التخطيط

تفكيك المهام ووضع الاستراتيجية.

نمط ReAct

دورة الاستدلال + التنفيذ:

1Thought: What do I need to do?
2Action: Which tool should I use?
3Action Input: Tool parameters
4Observation: Tool result
5... (repeat)
6Final Answer: Final response

مثال ReAct

1Question: "What is Tesla's current stock price and how much has it changed in the last month?"
2
3Thought: I need to find Tesla's stock price.
4Action: stock_price
5Action Input: {"symbol": "TSLA"}
6Observation: Current price: $248.50
7
8Thought: Now I need to calculate the 1-month change.
9Action: stock_history  
10Action Input: {"symbol": "TSLA", "period": "1mo"}
11Observation: 1 month ago: $235.20, Change: +5.65%
12
13Thought: I can now answer the question.
14Final Answer: Tesla's current stock price is $248.50. 
15It has shown a 5.65% increase in the last month.

تطوير العملاء باستخدام LangChain

عميل أساسي

1from langchain.agents import create_react_agent, AgentExecutor
2from langchain.tools import Tool
3from langchain_openai import ChatOpenAI
4from langchain import hub
5
6# LLM
7llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
8
9# Tools
10def search_web(query: str) -> str:
11    """Searches the web"""
12    # Web search implementation
13    return f"Search results for: {query}"
14
15def calculate(expression: str) -> str:
16    """Performs mathematical calculation"""
17    return str(eval(expression))
18
19tools = [
20    Tool(name="search", func=search_web, description="Searches the web"),
21    Tool(name="calculator", func=calculate, description="Performs math calculations"),
22]
23
24# Prompt
25prompt = hub.pull("hwchase17/react")
26
27# Agent
28agent = create_react_agent(llm, tools, prompt)
29agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
30
31# Run
32result = agent_executor.invoke({"input": "What is the world population in 2024?"})

أدوات مخصصة

1from langchain.tools import BaseTool
2from pydantic import BaseModel, Field
3
4class WeatherInput(BaseModel):
5    city: str = Field(description="City name")
6    unit: str = Field(default="celsius", description="Temperature unit")
7
8class WeatherTool(BaseTool):
9    name = "weather"
10    description = "Fetches weather information"
11    args_schema = WeatherInput
12    
13    def _run(self, city: str, unit: str = "celsius") -> str:
14        # API call
15        weather_data = fetch_weather(city)
16        return f"{city}: {weather_data['temp']}°{'C' if unit == 'celsius' else 'F'}"
17    
18    async def _arun(self, city: str, unit: str = "celsius") -> str:
19        return await async_fetch_weather(city)
20## أنظمة الذاكرة
21
22### ذاكرة تخزين المحادثة
23
24```python
25from langchain.memory import ConversationBufferMemory
26
27memory = ConversationBufferMemory(
28    memory_key="chat_history",
29    return_messages=True
30)
31
32# Add to Agent
33agent_executor = AgentExecutor(
34    agent=agent,
35    tools=tools,
36    memory=memory,
37    verbose=True
38)

ذاكرة الملخص

ملخص للمحادثات الطويلة:

1from langchain.memory import ConversationSummaryMemory
2
3memory = ConversationSummaryMemory(
4    llm=llm,
5    memory_key="chat_history"
6)

ذاكرة مخزن المتجهات

ذاكرة طويلة المدى ببحث دلالي:

1from langchain.memory import VectorStoreRetrieverMemory
2from langchain.vectorstores import Chroma
3
4vectorstore = Chroma(embedding_function=embeddings)
5retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
6
7memory = VectorStoreRetrieverMemory(retriever=retriever)

أنظمة الوكلاء المتعددين

نمط المشرف

1from langgraph.graph import StateGraph, END
2
3def supervisor_agent(state):
4    """Supervisor distributing tasks"""
5    task = state["task"]
6    
7    if "research" in task:
8        return "researcher"
9    elif "write" in task:
10        return "writer"
11    else:
12        return "general"
13
14def researcher_agent(state):
15    """Agent conducting research"""
16    # Research logic
17    return {"research_results": results}
18
19def writer_agent(state):
20    """Agent writing content"""
21    # Writing logic
22    return {"content": content}
23
24# Create Graph
25workflow = StateGraph(AgentState)
26workflow.add_node("supervisor", supervisor_agent)
27workflow.add_node("researcher", researcher_agent)
28workflow.add_node("writer", writer_agent)
29
30workflow.add_conditional_edges("supervisor", route_to_agent)
31workflow.add_edge("researcher", "supervisor")
32workflow.add_edge("writer", END)
33
34app = workflow.compile()

الوكلاء التعاونيون

1┌─────────────────────────────────────────────────────┐
2│                    Orchestrator                      │
3└───────────────────────┬─────────────────────────────┘
4                        │
5        ┌───────────────┼───────────────┐
6        │               │               │
7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐
8│   Researcher  │ │  Analyst  │ │   Writer    │
9│    Agent      │ │   Agent   │ │   Agent     │
10└───────────────┘ └───────────┘ └─────────────┘

استراتيجيات التخطيط

التخطيط والتنفيذ

1from langchain_experimental.plan_and_execute import (
2    PlanAndExecute,
3    load_agent_executor,
4    load_chat_planner
5)
6
7planner = load_chat_planner(llm)
8executor = load_agent_executor(llm, tools, verbose=True)
9
10agent = PlanAndExecute(planner=planner, executor=executor)
11
12result = agent.run("Research and create a draft for a blog post")

تقسيم المهام

1Main Task: "Analyze products for an e-commerce site"
2
3Sub-Tasks:
41. Research competitor products
52. Perform price comparison
63. Analyze customer reviews
74. Generate report
8
9Each sub-task is handled by an agent.
10## التعامل مع الأخطاء والمحاولة من جديد
11
12```python
13from langchain.callbacks import get_openai_callback
14
15class RobustAgentExecutor:
16    def __init__(self, agent_executor, max_retries=3):
17        self.executor = agent_executor
18        self.max_retries = max_retries
19    
20    def invoke(self, input_dict):
21        for attempt in range(self.max_retries):
22            try:
23                with get_openai_callback() as cb:
24                    result = self.executor.invoke(input_dict)
25                    
26                    # Token limit check
27                    if cb.total_tokens > 10000:
28                        self.summarize_memory()
29                    
30                    return result
31                    
32            except Exception as e:
33                if attempt == self.max_retries - 1:
34                    raise
35                
36                # Strategy based on Error
37                if "rate_limit" in str(e):
38                    time.sleep(60)
39                elif "context_length" in str(e):
40                    self.truncate_context()
41                else:
42                    time.sleep(2 ** attempt)

البث والمعالجة في الوقت الحقيقي

1from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
2
3llm = ChatOpenAI(
4    streaming=True,
5    callbacks=[StreamingStdOutCallbackHandler()]
6)
7
8async def stream_agent_response(query):
9    async for event in agent_executor.astream_events(
10        {"input": query},
11        version="v1"
12    ):
13        kind = event["event"]
14        if kind == "on_chat_model_stream":
15            print(event["data"]["chunk"].content, end="")
16        elif kind == "on_tool_start":
17            print(f"\n🔧 Using tool: {event['name']}")

اعتبارات الإنتاج

السجلات والمراقبة

1import logging
2
3class AgentLogger:
4    def __init__(self):
5        self.logger = logging.getLogger("agent")
6    
7    def log_action(self, action, input_data, output):
8        self.logger.info({
9            "action": action,
10            "input": input_data,
11            "output_length": len(str(output)),
12            "timestamp": datetime.now().isoformat()
13        })
14    
15    def log_error(self, error, context):
16        self.logger.error({
17            "error": str(error),
18            "context": context
19        })

إدارة التكلفة

1def estimate_cost(agent_run):
2    # Cost per token (GPT)
3    input_cost = agent_run.input_tokens * 0.00003
4    output_cost = agent_run.output_tokens * 0.00006
5    
6    return input_cost + output_cost
7
8# Budget limit
9MAX_BUDGET_PER_REQUEST = 0.50  # $0.50
10
11if estimate_cost(run) > MAX_BUDGET_PER_REQUEST:
12    raise BudgetExceededError()

الخلاصة

تمكّن AI Agents التطبيقات من العمل بذكاء ذاتي وبشكل استباقي. تسهّل أطر العمل مثل LangChain و LangGraph تطوير أنظمة عملاء جاهزة للإنتاج.

في Veni AI، نقوم بتطوير حلول عملاء للمؤسسات.