AI Agent Architectuur: Autonome Agents en LangChain Framework
AI Agents zijn artificiële-intelligentiesystemen die zelfstandig beslissingen kunnen nemen, tools kunnen gebruiken en complexe taken stap voor stap kunnen oplossen. In deze gids bekijken we agentarchitecturen en hun implementaties.
Wat is een Agent?
Een eenvoudige vergelijking tussen een chatbot en een 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."
Componenten van een Agent
1. LLM (Brein)
Besluitvorming en redenering.
2. Tools
Interactie met de buitenwereld:
- Web search
- API calls
- Database queries
- Code execution
3. Memory
Gespreksgeschiedenis en informatieopslag.
4. Planning
Taakdecompositie en strategie.
ReAct Pattern
Redeneren + Handelen-cyclus:
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 Voorbeeld
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.
Agentontwikkeling met 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## Geheugensystemen 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
Samenvatting voor lange conversaties:
1from langchain.memory import ConversationSummaryMemory 2 3memory = ConversationSummaryMemory( 4 llm=llm, 5 memory_key="chat_history" 6)
Vector Store Memory
Langetermijngeheugen met semantisch zoeken:
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 Systemen
Supervisorpatroon
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()
Samenwerkende Agenten
1┌─────────────────────────────────────────────────────┐ 2│ Orchestrator │ 3└───────────────────────┬─────────────────────────────┘ 4 │ 5 ┌───────────────┼───────────────┐ 6 │ │ │ 7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐ 8│ Researcher │ │ Analyst │ │ Writer │ 9│ Agent │ │ Agent │ │ Agent │ 10└───────────────┘ └───────────┘ └─────────────┘
Planningsstrategieën
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")
Taakdecompositie
1Hoofdtaak: "Analyze products for an e-commerce site" 2 3Subtaken: 41. Research competitor products 52. Perform price comparison 63. Analyze customer reviews 74. Generate report 8 9Elke subtaak wordt afgehandeld door een agent. 10## Foutafhandeling en Retry 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 en Realtime
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']}")
Overwegingen voor Productie
Logging en Monitoring
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 })
Kostenbeheer
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()
Conclusie
AI Agents maken autonome en proactieve toepassingen van kunstmatige intelligentie mogelijk. Frameworks zoals LangChain en LangGraph maken het eenvoudig om productierijpe agent-systemen te ontwikkelen.
Bij Veni AI ontwikkelen we enterprise agent-oplossingen.