Αρχιτεκτονική AI Agent: Αυτόνομοι Agents και LangChain Framework

Οι AI Agents είναι συστήματα τεχνητής νοημοσύνης που μπορούν να λαμβάνουν αποφάσεις μόνα τους, να χρησιμοποιούν εργαλεία και να επιλύουν πολύπλοκες εργασίες βήμα προς βήμα. Σε αυτόν τον οδηγό, εξετάζουμε αρχιτεκτονικές agents και τις υλοποιήσεις τους.

Τι είναι ένας Agent;

Μια απλή σύγκριση μεταξύ ενός chatbot και ενός 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."

Συνιστώσες Agent

1. LLM (Brain)

Λήψη αποφάσεων και συλλογισμός.

2. Εργαλεία (Tools)

Αλληλεπίδραση με τον έξω κόσμο:

• Web search
• API calls
• Database queries
• Code execution

3. Μνήμη (Memory)

Ιστορικό συνομιλιών και αποθήκευση πληροφοριών.

4. Σχεδιασμός (Planning)

Ανάλυση εργασιών και στρατηγική.

ReAct Pattern

Κύκλος Reasoning + Acting:

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.

Ανάπτυξη Agent με LangChain

Basic Agent

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?"})

Custom Tools

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### Conversation Buffer Memory
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)

Summary Memory

Περίληψη για μεγάλες συνομιλίες:

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

Vector Store Memory

Μακροχρόνια μνήμη με semantic search:

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)

Multi-Agent Systems

Supervisor Pattern

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()

Collaborative Agents

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

Στρατηγικές Σχεδιασμού

Plan-and-Execute

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")

Task Decomposition

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)

Ροή Δεδομένων (Streaming) και Πραγματικός Χρόνος

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 επιτρέπουν αυτόνομες και προδραστικές εφαρμογές τεχνητής νοημοσύνης. Τα frameworks όπως τα LangChain και LangGraph διευκολύνουν την ανάπτυξη agent συστημάτων έτοιμων για παραγωγή.

Στη Veni AI, αναπτύσσουμε λύσεις enterprise agents.