Ein AI-Agent ist ein LLM mit Tools: Der Agent denkt, entscheidet wann welche Tools zu nutzen, und führt Aktionen aus. Ein Chatbot antwortet auf Fragen. Ein Agent antwortet auf Aufgaben.
Das Problem: Agenten bauen ist komplex. Tools müssen registered werden, Fehler müssen gehandhabt werden, die Loop (denk → tool call → ergebnis → denk weiter) muss orchestriert werden.
Lösung 1: LangChain oder CrewAI — fertige Frameworks. Lösung 2: Selbst bauen. Wir zeigen beide, mit konkretem Beispiel (Dokumenten-Analyse-Agent). Und: Wann welche sinnvoll ist.
Was ist ein AI Agent wirklich?
Ein Agent ist eine Loop:
- **LLM-Call**: "Hier ist die Aufgabe, hier sind deine Tools. Was tust du?"
- **Tool Selection**: LLM antwortet: "Ich rufe Tool X auf mit Args Y"
- **Tool Execution**: Dein Code führt Tool X(Y) aus
- **Loop**: Ergebnis an LLM zurück, rinse & repeat bis LLM sagt "Fertig"
Beispiel: "Analysiere dieses PDF und antworte auf Fragen dazu" - Agent denkt: "Ich muss zuerst das PDF lesen" - Agent calls Tool `load_pdf` - Tool gibt zurück: "PDF gelesen, 50 Seiten" - Agent denkt: "Jetzt analysiere ich..." - Agent calls Tool `analyze_document` - Tool gibt zurück: "Analyse fertig: Thema ist ..." - Agent denkt: "Fertig, gebe Antwort"
Die Kunst ist: Wie kriegt die LLM zu entscheiden, wann Tools zu benutzen? Das ist wo Frameworks helfen (oder auch nicht).
Einfacher Custom Agent (selbst gebaut)
Zuerst zeigen wir wie einfach es sein kann.
Tools registrieren: ```python import json from typing import Any, Callable
class AgentToolkit: def __init__(self): self.tools: dict[str, Callable] = {} def register(self, name: str, description: str): """Decorator to register a tool""" def decorator(func: Callable) -> Callable: self.tools[name] = { "func": func, "description": description, "signature": func.__annotations__ } return func return decorator def get_tools_schema(self): """Return tools in format for LLM""" return { name: { "description": info["description"], "params": {k: str(v) for k, v in info["signature"].items() if k != "return"} } for name, info in self.tools.items() }
tools = AgentToolkit()
@tools.register("load_document", "Load and read a document file") def load_document(filepath: str) -> str: """Load document content""" with open(filepath, 'r') as f: return f.read()[:5000] # First 5000 chars
@tools.register("search_document", "Search for text in loaded document") def search_document(query: str) -> str: """Search in document (in real life, would search the loaded doc)""" return f"Found 3 matches for '{query}'..." ```
Simple Agent Loop: ```python from anthropic import Anthropic import json
client = Anthropic()
def run_agent(user_task: str, max_iterations: int = 10): messages = [ {"role": "user", "content": user_task} ] tools_schema = tools.get_tools_schema() tools_str = json.dumps(tools_schema, indent=2) system_prompt = f"""You are a helpful document analysis agent. You have access to these tools: {tools_str}
When you need to use a tool, respond with: <tool_call> <name>tool_name</name> <params>{{"param": "value"}}</params> </tool_call>
When you have the answer, respond normally with the final answer.""" for iteration in range(max_iterations): # Ask LLM what to do response = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1000, system=system_prompt, messages=messages ) assistant_message = response.content[0].text messages.append({"role": "assistant", "content": assistant_message}) # Check if LLM wants to use a tool if "<tool_call>" not in assistant_message: # LLM is done, return answer return assistant_message # Parse tool call import re tool_match = re.search( r'<tool_call>\n<name>(\w+)</name>\n<params>({[^}]+})</params>\n</tool_call>', assistant_message ) if not tool_match: return "Error: Could not parse tool call" tool_name = tool_match.group(1) params_str = tool_match.group(2) params = json.loads(params_str) # Execute tool if tool_name not in tools.tools: result = f"Error: Unknown tool '{tool_name}'" else: try: result = tools.tools[tool_name]["func"](**params) except Exception as e: result = f"Error executing tool: {str(e)}" # Add result to messages messages.append({"role": "user", "content": f"Tool result: {result}"}) return "Max iterations reached"
# Usage answer = run_agent("Load document.pdf and find all mentions of 'budget'") print(answer) ```
Eigenschaften: - ✓ Vollständig unter Kontrolle - ✓ Einfach zu debuggen (you see every step) - ✓ Minimal Dependencies (nur Anthropic SDK) - ✗ Keine Error Recovery (wenn Tool crasht, Loop bricht) - ✗ Keine Parallele Tool Calls - ✗ Kein Tracing/Logging out-of-the-box
Für einfache Agenten (1-3 Tools, straightforward Logic): Ausreichend.
LangChain ReAct Agent
LangChain bringt eine bewährte Agent-Architektur: ReAct (Reasoning + Acting).
Setup (Requirements: `pip install langchain anthropic`): ```python from langchain.agents import tool, initialize_agent, AgentType from langchain.chat_models import ChatAnthropic from langchain.tools import Tool
model = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)
# Tools mit Decorator @tool def load_document(filepath: str) -> str: """Load and return document content (first 5000 chars)""" try: with open(filepath, 'r') as f: return f.read()[:5000] except FileNotFoundError: return f"Error: File {filepath} not found"
@tool def search_document(query: str, document_content: str) -> str: """Search for text in document""" matches = [line for line in document_content.split("\n") if query.lower() in line.lower()] if not matches: return f"No matches found for '{query}'" return f"Found {len(matches)} matches:\n" + "\n".join(matches[:5])
@tool def analyze_text(text: str) -> str: """Analyze text and extract key insights""" # In real app: use LLM for analysis, DB queries, etc. return f"Analysis: '{text[:100]}...' seems to be about business topics"
tools = [load_document, search_document, analyze_text]
# Create agent agent = initialize_agent( tools, model, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True, # Print thinking steps max_iterations=10, )
# Run result = agent.run("Load report.pdf and find all budget mentions") print(result) ```
LangChain bringt: - ✓ Standardisierte Tool-Definition (mit Decorator) - ✓ Built-in Error Handling - ✓ Multiple Agent Types (ReAct, OpenAI Functions, etc.) - ✓ Tracing/Callback System - ✓ Integration mit vielen LLM Providern - ✗ Learning Curve (LangChain hat viele Konzepte) - ✗ Performance Overhead (extra Wrapper) - ✗ Locking dich in LangChain-Ecosystem ein
Aber Vorsicht: LangChain ist viel und ändert sich. Code von 2024 läuft vielleicht 2026 nicht mehr ohne Anpassungen.
CrewAI: Multi-Agent Orchestration
CrewAI ist neuere, spezialisiert auf Multi-Agent Szenarien (mehrere Agenten arbeiten zusammen).
Setup (Requirements: `pip install crewai`): ```python from crewai import Agent, Task, Crew from crewai.tools import tool
@tool def load_document(filepath: str) -> str: """Load document""" with open(filepath, 'r') as f: return f.read()[:5000]
@tool def search_in_text(query: str, text: str) -> str: """Search for text""" matches = [line for line in text.split("\n") if query in line] return f"Found {len(matches)} matches"
# Agent 1: Document Reader reader_agent = Agent( role="Document Reader", goal="Load and understand documents", backstory="You are expert at reading and understanding documents.", tools=[load_document], verbose=True, )
# Agent 2: Analyst analyst_agent = Agent( role="Data Analyst", goal="Analyze documents and find insights", backstory="You are expert analyst.", tools=[search_in_text], verbose=True, )
# Tasks task1 = Task( description="Load report.pdf", agent=reader_agent, expected_output="Content of the document" )
task2 = Task( description="Find all mentions of budget in the loaded content", agent=analyst_agent, expected_output="List of budget-related findings" )
# Crew: Koordiniere beide Agenten crew = Crew( agents=[reader_agent, analyst_agent], tasks=[task1, task2], verbose=True, )
result = crew.kickoff() print(result) ```
CrewAI Stärken: - ✓ Multi-Agent Coordination out-of-the-box - ✓ Task-basiert (natural für komplexe Workflows) - ✓ Neuere, aktiveres Projekt als LangChain - ✗ Noch nicht stabil (API ändert sich) - ✗ Weniger Dokumentation - ✗ Weniger integrations
CrewAI ist Gut wenn: Mehrere Agenten zusammenarbeiten müssen (Analyst + Coder + Reviewer).
Vergleich: Custom vs. LangChain vs. CrewAI
| Aspekt | Custom | LangChain | CrewAI | |--------|--------|-----------|--------| | Setup-Zeit | 30 min | 1 hour | 1-2 hours | | Komplexität | Niedrig | Mittel | Mittel-Hoch | | Tool-Definition | Einfach | Standardisiert | Standardisiert | | Error-Handling | Du musst selbst | Built-in | Built-in | | Multi-Agent | ✗ | Möglich aber awkward | ✓ ✓ | | Production-Ready | Für einfache Fälle | Ja | Jetzt | | Debugging | Einfach | Verbose Output | Einfach | | Lock-in Risiko | Keiner | Mittel | Hoch (junge Projekt) | | Dependency Count | 1 (anthropic) | 20+ (LangChain ecosystem) | 10+ | | Performance | Schnell | Mittel (overhead) | Mittel |
Wann Custom? - Sehr einfache Agents (1-3 Tools) - Maximale Control wichtig - Minimale Dependencies erwünscht - Schnelle Iteration nötig
Wann LangChain? - Single Agent mit mehreren Tools - Integration mit vielen LLM Providern nötig - Du willst Standardisierung über dein Team - Production-Reife wichtig
Wann CrewAI? - Multi-Agent Systeme - Task-orchestration zentral - Du magst den Design-Ansatz
Praxis-Beispiel: Document Analysis Agent mit eigener Lösung
Kompletteres Beispiel mit Error Handling, Logging, etc.
```python import json import logging from typing import Any, Optional from anthropic import Anthropic
logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__)
class DocumentAgent: def __init__(self, model: str = "claude-3-5-sonnet-20241022"): self.client = Anthropic() self.model = model self.documents = {} # Cache loaded documents self.tools = { "load_document": self._load_document, "search_document": self._search_document, "extract_section": self._extract_section, "summarize": self._summarize, } def _load_document(self, filepath: str) -> str: if filepath not in self.documents: try: with open(filepath, 'r') as f: content = f.read() self.documents[filepath] = content logger.info(f"Loaded {filepath}: {len(content)} chars") except Exception as e: return f"Error loading {filepath}: {str(e)}" return f"Document {filepath} loaded ({len(self.documents[filepath])} chars)" def _search_document(self, query: str, filepath: str) -> str: if filepath not in self.documents: return f"Document {filepath} not loaded" doc = self.documents[filepath] lines = [line for line in doc.split("\n") if query.lower() in line.lower()] if not lines: return f"No matches for '{query}'" return f"Found {len(lines)} matches:\n" + "\n".join(lines[:3]) def _extract_section(self, heading: str, filepath: str) -> str: if filepath not in self.documents: return f"Document {filepath} not loaded" doc = self.documents[filepath] # Simple section extraction (in real: use regex) idx = doc.find(heading) if idx == -1: return f"Section '{heading}' not found" return doc[idx:idx+2000] def _summarize(self, filepath: str) -> str: if filepath not in self.documents: return f"Document {filepath} not loaded" # Would call LLM to summarize return "Summary: This document discusses important topics." def run(self, task: str, max_iterations: int = 10) -> str: messages = [{"role": "user", "content": task}] tools_descriptions = { "load_document": "Load a document file and make it available for analysis", "search_document": "Search for text in a loaded document", "extract_section": "Extract a specific section from document", "summarize": "Summarize a loaded document", } system = f"""You are a document analysis agent. Available tools: {json.dumps(tools_descriptions, indent=2)}
When using a tool, respond with: <tool_call> <name>tool_name</name> <params>{{"key": "value"}}</params> </tool_call>""" for i in range(max_iterations): logger.info(f"Iteration {i+1}") response = self.client.messages.create( model=self.model, max_tokens=2000, system=system, messages=messages ) assistant_msg = response.content[0].text messages.append({"role": "assistant", "content": assistant_msg}) if "<tool_call>" not in assistant_msg: logger.info("Agent finished") return assistant_msg # Parse and execute tool import re match = re.search( r'<tool_call>\n<name>(\w+)</name>\n<params>({.*?})</params>\n</tool_call>', assistant_msg, re.DOTALL ) if not match: messages.append({"role": "user", "content": "Error: Invalid tool call format"}) continue tool_name = match.group(1) params = json.loads(match.group(2)) if tool_name not in self.tools: result = f"Error: Unknown tool {tool_name}" else: try: result = self.tools[tool_name](**params) logger.info(f"Tool {tool_name} result: {result[:100]}...") except Exception as e: result = f"Error: {str(e)}" messages.append({"role": "user", "content": f"Tool output: {result}"}) return "Max iterations reached"
# Usage agent = DocumentAgent() result = agent.run("Load report.pdf, find all budget sections, and summarize them") print(result) ```
Key Features: - Caching of loaded documents - Error Handling für jeden Tool-Call - Logging für Debugging - Max iterations safeguard - Clean tool abstraction
Integration in Django
Ein Agent sollte nicht in einem HTTP-Request laufen (zu lange). Stattdessen: Async Task mit Celery.
```python # tasks.py from celery import shared_task from document_agent import DocumentAgent
@shared_task def analyze_document_async(task_description: str, user_id: int): agent = DocumentAgent() result = agent.run(task_description) # Store result in DB from myapp.models import AnalysisResult AnalysisResult.objects.create( user_id=user_id, task=task_description, result=result ) return result
# views.py from django.http import JsonResponse from .tasks import analyze_document_async
def start_analysis(request): task_description = request.POST.get("task") # Trigger async task celery_task = analyze_document_async.delay( task_description, request.user.id ) return JsonResponse({ "task_id": celery_task.id, "status": "processing" })
def get_analysis_result(request, task_id): from celery.result import AsyncResult task = AsyncResult(task_id) if task.ready(): return JsonResponse({"status": "done", "result": task.result}) else: return JsonResponse({"status": "processing"}) ```
Wichtige Überlegungen
1. Infinite Loops: Ein fehlerhafte Agent kann sich in einer Loop verfangen (Tool A calls Tool B calls Tool A ...). Immer `max_iterations` setzen.
2. Cost Control: Jeder LLM-Call kostet. Für Testing: GPT-3.5 nutzen (günstiger) bevor Production.
3. Determinism: LLMs sind nicht deterministisch. Einen Agent zweimal laufen zu lassen kann unterschiedliche Tools aufrufen. Das ist OK, aber bedenk es.
4. Tool Definitions: Je präziser die Tool-Beschreibung, desto besser wählt die LLM. "Load document" vs "Load a text file and return its full content for analysis". Latter ist besser.
5. Feedback Loop: Nach 2-3 Iterationen: Wenn Agent noch nicht fertig, wahrscheinlich stuck. Error Messages an Agent helfen ("Tool returned error: ...").
Fazit
AI-Agenten sind mächtig, aber Komplexität ist real. Die Wahl zwischen Custom, LangChain und CrewAI hängt von deinem Szenario ab.
Faustregeln: - Einfacher Agent (1-3 Tools, straightforward Logic) → Custom - Einzel-Agent, viele Tools → LangChain - Multi-Agent Systeme → CrewAI
Wir empfehlen: Starte mit Custom (Du verstehst die Loops). Wenn es komplexer wird, migrate zu LangChain. CrewAI später wenn du Multi-Agent Orchestration brauchst.
Wenn du einen Agent für ein spezifisches Business-Problem bauen willst und nicht sicher bist über Architektur oder Skalierung: Unser KI-Readiness-Check schaut sich deine Use-Case an und zeigt dir, welcher Ansatz passt und welche Tools zu wählen sind.