Architektur von AI Agents: Autonome Agents und das LangChain-Framework
AI Agents sind künstliche Intelligenzsysteme, die eigenständig Entscheidungen treffen, Tools verwenden und komplexe Aufgaben Schritt für Schritt lösen können. In diesem Leitfaden untersuchen wir Agent-Architekturen und deren Implementierungen.
Was ist ein Agent?
Ein einfacher Vergleich zwischen einem Chatbot und einem 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."
Komponenten eines Agents
1. LLM (Gehirn)
Entscheidungsfindung und logisches Denken.
2. Tools
Interaktion mit der Außenwelt:
- Websuche
- API-Aufrufe
- Datenbankabfragen
- Codeausführung
3. Memory
Gesprächsverlauf und Informationsspeicherung.
4. Planning
Aufgabenzerlegung und Strategie.
ReAct Pattern
Zyklus aus 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 Beispiel
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-Entwicklung mit 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## Speichersysteme 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
Zusammenfassung für lange Gespräche:
1from langchain.memory import ConversationSummaryMemory 2 3memory = ConversationSummaryMemory( 4 llm=llm, 5 memory_key="chat_history" 6)
Vector Store Memory
Langzeitspeicher mit semantischer Suche:
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-Agenten-Systeme
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()
Kollaborative Agenten
1┌─────────────────────────────────────────────────────┐ 2│ Orchestrator │ 3└───────────────────────┬─────────────────────────────┘ 4 │ 5 ┌───────────────┼───────────────┐ 6 │ │ │ 7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐ 8│ Researcher │ │ Analyst │ │ Writer │ 9│ Agent │ │ Agent │ │ Agent │ 10└───────────────┘ └───────────┘ └─────────────┘
Planungsstrategien
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")
Aufgabenzerlegung
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## Fehlerbehandlung und Wiederholungsmechanismen 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 und Echtzeit
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']}")
Überlegungen für den Produktionseinsatz
Logging und 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 })
Kostenmanagement
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()
Fazit
AI Agents ermöglichen autonome und proaktive Anwendungen künstlicher Intelligenz. Frameworks wie LangChain und LangGraph machen es einfach, produktionsreife Agentensysteme zu entwickeln.
Bei Veni AI entwickeln wir Agentenlösungen für Unternehmen.