Ok, der User will ein Agentensystem für seine Werbeagentur. Es soll: 1. Tool-unabhängig sein (also nicht von bestimmten APIs oder Plattformen abhängen) 2. Komplexes Wissen/Erinnerung/Kontext-System haben, das über RAG (Retrieval-Augmented Generation) und OKF (was auch immer das ist - vielleicht "Open Knowledge Framework" oder ähnliches?) hinausgeht 3. Schnell und kostengünstig aufgebaut werden Erstmal verstehe ich die Anforderungen: - **Tool-unabhängig**: Das System sollte modular sein, sodass Agenten verschiedene Tools nutzen können, ohne an bestimmte Anbieter gebunden zu sein. Vielleicht über Adapter-Pattern oder Standard-Schnittstellen (z.B. REST, gRPC). - **Komplexes Wissen/Erinnerung/Kontext**: - Wissen: Fakten, Daten, Branchenspezifisches (z.B. Werbestrategien, Kundenprofile) - Erinnerung: Langzeitgedächtnis für Interaktionen (z.B. "Kunde X mag blau und hat Budget Y") - Kontext: Aktueller Zustand (z.B. "Wir sind in Phase 2 der Kampagne für Kunde Z") - Über RAG hinaus: RAG ist gut für statisches Wissen, aber hier braucht man dynamische Erinnerung und Kontext. Vielleicht eine Kombination aus: - Vektordatenbank für semantische Suche (RAG) - Graphdatenbank für Beziehungen (z.B. "Kunde -> Kampagne -> Zielgruppe") - Zeitreihen- oder Event-Sourcing für Erinnerung (z.B. "Letzte Interaktion war am...") - Arbeitsgedächtnis für aktuellen Kontext (z.B. in einer Session) - **Schnell & kostengünstig**: Keine jahrzehntelange Entwicklung, keine teuren Enterprise-Tools. Vielleicht Open-Source-Komponenten, Cloud-Dienste mit Pay-as-you-go, oder sogar lokal gehostet. ### Mögliche Architektur: 1. **Agenten-Framework**: - Basis: Ein Framework wie [AutoGen](https://github.com/microsoft/autogen) (Python), [LangChain](https://github.com/langchain-ai/langchain) oder [CrewAI](https://github.com/joaomdmoura/crewAI). Die sind Open Source und bieten schon viel Funktionalität für Multi-Agenten-Systeme. - Alternativ: Eigenbau mit Python + FastAPI für die Orchestrierung. 2. **Wissenssystem**: - **RAG**: Für statisches Wissen (z.B. Dokumente, FAQs). Kann mit [FAISS](https://github.com/facebookresearch/faiss) oder [Weaviate](https://weaviate.io/) (kostenlose Version) umgesetzt werden. - **Erinnerung**: - Kurzfristig: In-Memory (z.B. Redis für Session-Daten) - Langfristig: Graphdatenbank wie [Neo4j](https://neo4j.com/) (kostenlose Community Edition) oder [Dgraph](https://dgraph.io/). Damit kann man Beziehungen zwischen Entitäten speichern (z.B. "Agent A hat mit Kunde B über Projekt C gesprochen"). - Alternativ: Zeitstempel-basierte Datenbank wie [TimescaleDB](https://www.timescale.com/) für zeitliche Erinnerungen. - **Kontext**: - Arbeitsgedächtnis: Ein einfacher Key-Value-Store (Redis) oder sogar ein Dictionary in Python für die aktuelle Session. - Für komplexere Fälle: Ein State-Management-System (z.B. [Apache Airflow](https://airflow.apache.org/) für Workflows, aber vielleicht Overkill). 3. **Tool-Unabhängigkeit**: - Definiere eine Standard-Schnittstelle für Tools (z.B. REST API mit festem Schema). - Nutze Adapter für verschiedene Tools (z.B. ein Adapter für Google Ads, einer für Facebook Ads, etc.). Jeder Adapter übersetzt die Standard-Schnittstelle in die API des Tools. - Beispiel: ```python class ToolAdapter: def execute(self, action: str, params: dict) -> dict: pass class GoogleAdsAdapter(ToolAdapter): def execute(self, action, params): # Übersetzt action/params in Google Ads API Aufruf pass ``` 4. **Agenten-Design**: - Spezialisierte Agenten für verschiedene Aufgaben: - **Wissensagent**: Verwendet RAG + GraphDB für Faktenabfrage. - **Erinnerungsagent**: Greift auf die GraphDB/Zeitreihendaten zu für historische Daten. - **Kreativagent**: Nutzt LLMs für Texterstellung (z.B. über OpenAI, aber auch lokal mit [Ollama](https://ollama.ai/) für Kostenersparnis). - **Orchestrierungsagent**: Koordiniert die anderen Agenten und hält den Kontext. - Jeder Agent hat Zugriff auf: - Das Wissenssystem (RAG + Graph) - Das Erinnerungssystem (GraphDB + Events) - Das Kontextsystem (Arbeitsgedächtnis) 5. **Datenfluss**: - Ein Nutzer stellt eine Anfrage (z.B. "Erstelle eine Kampagne für Kunde X"). - Der Orchestrierungsagent: 1. Lädt den Kontext (z.B. "Aktueller Kunde: X, Phase: Planung"). 2. Frag den Erinnerungsagenten: "Was wissen wir über Kunde X?" 3. Frag den Wissensagenten: "Welche Strategien passen zu Zielgruppe Y?" 4. Der Kreativagent erstellt Entwürfe. 5. Der Tool-Agent interagiert mit externen Tools (z.B. Google Ads API) um die Kampagne zu erstellen. 6. **Kostengünstig & Schnell**: - **Infrastruktur**: - Lokale Entwicklung: Alles auf einem Rechner mit Docker (z.B. Neo4j, FAISS, Redis in Containern). - Produktion: Günstige Cloud-Anbieter wie [Hetzner Cloud](https://www.hetzner.com/cloud) oder [Oracle Cloud Free Tier](https://www.oracle.com/cloud/free/). - LLMs: Nutze lokale Modelle (z.B. [Mistral-7B](https://mistral.ai/) über Ollama) statt teurer API-Aufrufe. - **Daten**: - Beginne mit kleinen, relevanten Datensätzen (z.B. eigene Kundenakten, Branchenreports). - Nutze kostenlose Datenquellen (z.B. [Common Crawl](https://commoncrawl.org/), [Wikidata](https://www.wikidata.org/)). - **Entwicklung**: - Starte mit einem Minimal Viable Product (MVP): 1. Ein einfacher Agent, der RAG + Erinnerung (als JSON-Datei) nutzt. 2. Dann füge GraphDB hinzu. 3. Dann Tool-Adapter. - Nutze bestehende Bibliotheken wo möglich (z.B. LangChain für RAG, Neo4j Python Driver für GraphDB). 7. **Beispiel-Stack**: - **Backend**: FastAPI (Python) für die API. - **Datenbanken**: - Vektoren: FAISS (lokal) oder Weaviate (Cloud/self-hosted). - Graph: Neo4j (Community Edition). - Erinnerung/Events: SQLite oder PostgreSQL mit Timescale-Erweiterung. - Kontext: Redis (für Session-Daten). - **LLMs**: - Lokal: Ollama mit Mistral-7B oder Llama2-7B. - Cloud (falls nötig): OpenAI (teuer) oder [Together.ai](https://www.together.ai/) (günstiger). - **Agenten-Framework**: CrewAI oder AutoGen. - **Tool-Adapter**: Eigenbau (siehe oben). 8. **Schritt-für-Schritt-Plan**: **Phase 1: MVP (1-2 Wochen)** - Ziel: Ein Agent, der auf Basis von Dokumenten (RAG) und einfacher Erinnerung (JSON-Datei) antwortet. - Schritte: 1. Dokumente sammeln (z.B. eigene Werbetexte, Kundenbriefings). 2. Dokumente in Vektoren umwandeln (z.B. mit `sentence-transformers` + FAISS). 3. Einfaches Flask/FastAPI Backend, das: - Nutzeranfragen entgegennimmt. - RAG für Antworten nutzt. - Erinnerung als Dictionary in einer JSON-Datei speichert (z.B. `{"user_123": {"last_interaction": "...", "preferences": {...}}}`). 4. Einfacher Chat-Interface (z.B. mit Streamlit oder Gradio). **Phase 2: Erinnerung & Kontext (2-3 Wochen)** - Ziel: Dynamische Erinnerung und Kontext über Sessions hinweg. - Schritte: 1. Neo4j einrichten und Beispieldaten laden (z.B. Kunden, Kampagnen, Zielgruppen). 2. Erinnerungsagent implementieren: - Speichert jede Interaktion als Knoten in Neo4j (z.B. `(User)-[INTERACTED_AT {time: ...}]->(Agent)`). - Kann Fragen wie "Was war die letzte Kampagne für Kunde X?" beantworten. 3. Kontext-System: - Nutze Redis für Session-Daten (z.B. aktueller Kunde, aktuelle Kampagne). - Oder einfaches Python-Objekt, das zwischen Anfragen erhalten bleibt. **Phase 3: Tool-Integration (2-4 Wochen)** - Ziel: Agenten können externe Tools nutzen (z.B. Google Ads, Canva). - Schritte: 1. Standard-Schnittstelle für Tools definieren (z.B. `execute_tool(tool_name, action, params)`). 2. Adapter für 1-2 Tools implementieren (z.B. Google Ads API, Canva API). 3. Tool-Agenten erstellen, die diese Adapter nutzen. 4. Orchestrierungsagenten erweitern, um Tools aufzurufen (z.B. "Erstelle eine Anzeige in Google Ads"). **Phase 4: Skalierung & Optimierung (laufend)** - Ziel: System stabilisieren, Performance verbessern, mehr Agenten hinzufügen. - Schritte: 1. Monitoring einrichten (z.B. Prometheus + Grafana für Performance-Metriken). 2. Caching für häufige Anfragen (z.B. Redis Cache für RAG-Ergebnisse). 3. Mehr Agenten hinzufügen (z.B. Analysen-Agent für Kampagnenperformance). 4. Benutzerfeedback einarbeiten (z.B. "Hat diese Antwort geholfen?"). 9. **Kostenübersicht**: - **Infrastruktur**: - Neo4j Community: Kostenlos (self-hosted). - FAISS/Weaviate: Kostenlos (self-hosted). - Redis: Kostenlos (self-hosted oder günstige Cloud-Optionen). - LLMs: Lokal mit Ollama (kostenlos, braucht gute GPU für große Modelle). - Cloud-Server: z.B. Hetzner Cloud Server ab ~5€/Monat. - **Entwicklung**: - Open-Source-Tools: Keine Lizenzkosten. - API-Kosten: Falls externe LLMs/Tools genutzt werden (z.B. OpenAI), aber kann minimiert werden durch lokale Modelle. - **Gesamtkosten (MVP)**: - ~0-50€/Monat (je nach Cloud-Nutzung und API-Aufrufen). 10. **Risiken & Herausforderungen**: - **Datenqualität**: RAG und Erinnerung sind nur so gut wie die zugrundeliegenden Daten. Muss sicherstellen, dass die Daten aktuell und relevant sind. - **Skalierbarkeit**: FAISS und Neo4j skalieren gut, aber bei sehr großen Datenmengen könnte es Engpässe geben. - **Tool-Adapter**: Jedes neue Tool erfordert einen neuen Adapter. Kann aufwendig werden, wenn viele Tools unterstützt werden sollen. - **LLM-Kosten**: Falls auf Cloud-LLMs gesetzt wird, können die Kosten bei vielen Anfragen steigen. Lösung: Lokale Modelle oder Caching. 11. **Alternativen/Erweiterungen**: - **Wissen**: - Statt nur RAG: Nutze [LlamaIndex](https://github.com/run-llama/llama_index) für fortgeschrittene Indexierung (z.B. Hierarchien, Graphen). - Für strukturierte Daten: SQL-Datenbank (PostgreSQL) + Text-to-SQL (z.B. mit [SQLite + LLM](https://github.com/neonbjb/tortoise-orm)). - **Erinnerung**: - Nutze [Mem0](https://github.com/mem0ai/mem0) (Open-Source Langzeitgedächtnis für LLMs). - Oder baue ein eigenes System mit [Pinecone](https://www.pinecone.io/) (kostenlose Tier verfügbar) für Vektorspeicher + Metadaten. - **Kontext**: - Nutze [LangChain's ConversationBufferMemory](https://python.langchain.com/docs/modules/memory/) für einfachen Kontext. - Für komplexere Fälle: [Microsoft Semantic Kernel](https://github.com/microsoft/semantic-kernel) für State-Management. - **Agenten-Framework**: - [AutoGen](https://github.com/microsoft/autogen) ist sehr mächtig für Multi-Agenten-Systeme. - [CrewAI](https://github.com/joaomdmoura/crewAI) ist einfacher und gut für schnellere Prototypen. 12. **Empfohlener Tech-Stack für MVP**: | Komponente | Technologie | Kosten | Aufwand | |---------------------|---------------------------|--------------|---------| | Agenten-Framework | CrewAI oder AutoGen | Kostenlos | Mittel | | Wissensdatenbank | FAISS + Sentence-Transformers | Kostenlos | Niedrig | | Erinnerungsdatenbank| Neo4j (Community) | Kostenlos | Mittel | | Kontextspeicher | Redis oder JSON-Datei | Kostenlos | Niedrig | | LLMs | Ollama + Mistral-7B | Kostenlos* | Niedrig | | Backend | FastAPI | Kostenlos | Niedrig | | Frontend | Streamlit oder Gradio | Kostenlos | Niedrig | | Tool-Adapter | Eigenbau (Python) | Kostenlos | Hoch | *: Braucht GPU für gute Performance, aber kann auch auf CPU laufen (langsamer). 13. **Beispiel-Implementierung (Pseudocode)**: ```python # Hauptagent (Orchestrator) class OrchestratorAgent: def __init__(self): self.knowledge_agent = KnowledgeAgent() self.memory_agent = MemoryAgent() self.creative_agent = CreativeAgent() self.tool_agent = ToolAgent() self.context = {} # Aktueller Kontext (z.B. {"current_customer": "X"}) def handle_query(self, query): # 1. Kontext laden context = self._load_context() # 2. Erinnerung abfragen memory = self.memory_agent.retrieve(query, context) # 3. Wissen abfragen knowledge = self.knowledge_agent.retrieve(query, context) # 4. Kreativität nutzen draft = self.creative_agent.generate(query, context, memory, knowledge) # 5. Tools nutzen (falls nötig) if "erstelle anzeige" in query.lower(): result = self.tool_agent.execute("google_ads", "create_ad", {"draft": draft}) return result else: return draft # Wissensagent (RAG) class KnowledgeAgent: def __init__(self): self.vector_db = FAISS.load_local("faiss_index") self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2') def retrieve(self, query, context): # Embedding der Query query_embedding = self.embedding_model.encode(query) # Suche in Vektordatenbank results = self.vector_db.search(query_embedding, k=3) return results # Erinnerungsagent (Neo4j) class MemoryAgent: def __init__(self): self.driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password")) def retrieve(self, query, context): with self.driver.session() as session: # Beispiel: Finde letzte Interaktion mit aktuellem Kunden customer = context.get("current_customer") result = session.run( "MATCH (u:User {name: $customer})-[:INTERACTED]->(a:Agent) RETURN a.message ORDER BY a.timestamp DESC LIMIT 1", customer=customer ) return result.single()[0] # Tool-Agent class ToolAgent: def __init__(self): self.adapters = { "google_ads": GoogleAdsAdapter(), "canva": CanvaAdapter() } def execute(self, tool_name, action, params): adapter = self.adapters.get(tool_name) if adapter: return adapter.execute(action, params) else: raise ValueError(f"Tool {tool_name} nicht unterstützt") ``` 14. **Nächste Schritte für den User**: - **Woche 1**: 1. Dokumente sammeln (z.B. 10-20 PDFs/Word-Dokumente mit Werbetexten, Kundeninfos). 2. Lokale Umgebung einrichten: - Python 3.10+ - Docker (für Neo4j, Redis) - Ollama + Mistral-7B installieren (`ollama pull mistral`) 3. Einfaches RAG-System mit FAISS aufbauen (Tutorial: [FAISS + LangChain](https://python.langchain.com/docs/modules/data_connection/retrievers/vectorstore/faiss)). 4. Einfaches Frontend mit Streamlit bauen, das RAG-Abfragen ermöglicht. - **Woche 2**: 1. Neo4j lokal installieren und Beispieldaten laden (z.B. Kunden, Kampagnen). 2. Erinnerungsagent implementieren (einfache Cypher-Abfragen). 3. Kontext-System als Python-Dict oder Redis hinzufügen. - **Woche 3**: 1. Ersten Tool-Adapter implementieren (z.B. für Google Ads API). 2. Orchestrator-Agent bauen, der die anderen Agenten koordiniert. - **Woche 4**: 1. Testen mit echten Use Cases (z.B. "Erstelle eine Kampagne für Kunde X"). 2. Feedback einholen und iterieren. 15. **Langfristige Optimierungen**: - **Wissen**: - Nutze [Unstructured.io](https://www.unstructured.io/) für bessere Dokumentenverarbeitung (Tabellen, PDFs). - Füge eine Graph-basierte Wissensdatenbank hinzu (z.B. mit [Neo4j + LangChain](https://python.langchain.com/docs/modules/data_connection/graph)). - **Erinnerung**: - Implementiere ein "Memory Stream" System, das alle Interaktionen als Events speichert (Event Sourcing). - Nutze [Mem0](https://github.com/mem0ai/mem0) für automatische Gedächtnisverwaltung. - **Kontext**: - Nutze [LangChain's ConversationBufferWindowMemory](https://python.langchain.com/docs/modules/memory/types/buffer_window) für Sliding-Window-Kontext. - **Tools**: - Baue eine "Tool Registry", die automatisch neue Adapter lädt. - Nutze [Apache Airflow](https://airflow.apache.org/) für komplexe Workflows (z.B. "Kampagne erstellen → Anzeigen schalten → Performance tracken"). - **Performance**: - Caching für häufige RAG-Abfragen (Redis). - Asynchrone Verarbeitung für lange Laufzeiten (z.B. Celery + RabbitMQ). - **Sicherheit**: - Authentifizierung für Nutzer (z.B. JWT). - Berechtigungen für Agenten (z.B. "Agent A darf nur lesen, Agent B darf Tools nutzen"). 16. **Tools & Bibliotheken im Detail**: - **RAG**: - [LangChain](https://python.langchain.com/): Einfache Integration von RAG mit verschiedenen Vektordatenbanken. - [LlamaIndex](https://github.com/run-llama/llama_index): Fortgeschrittene Indexierung (z.B. für Tabellen, Graphen). - [FAISS](https://github.com/facebookresearch/faiss): Effiziente Vektorsuche (lokal). - [Weaviate](https://weaviate.io/): Vektordatenbank mit Graph-Funktionen (kostenlose Cloud-Version verfügbar). - **Graphdatenbanken**: - [Neo4j](https://neo4j.com/): Beliebteste GraphDB, gute Python-Bibliothek (`neo4j`). - [Dgraph](https://dgraph.io/): Skalierbar, gut für große Datenmengen. - **LLMs**: - [Ollama](https://ollama.ai/): Einfache lokale Ausführung von LLMs (z.B. Mistral, Llama2). - [Hugging Face Transformers](https://huggingface.co/docs/transformers/index): Für eigene Modelle. - [vLLM](https://github.com/vllm-project/vllm): Optimierte Inference für LLMs (falls eigene GPU-Server). - **Agenten-Frameworks**: - [AutoGen](https://github.com/microsoft/autogen): Microsoft's Framework für Multi-Agenten-Konversationen. - [CrewAI](https://github.com/joaomdmoura/crewAI): Einfacheres Framework, gut für Prototypen. - [LangGraph](https://github.com/langchain-ai/langgraph): Für komplexe Agenten-Workflows (von LangChain). - **Tool-Integration**: - [Apache Airflow](https://airflow.apache.org/): Für Workflow-Orchestrierung. - [Prefect](https://www.prefect.io/): Alternative zu Airflow, einfacher zu nutzen. - **Frontend**: - [Streamlit](https://streamlit.io/): Schnelles Prototyping für Web-Interfaces. - [Gradio](https://gradio.app/): Einfache UIs für ML-Modelle. - [FastAPI + React](https://fastapi.tiangolo.com/): Für professionellere Frontends. 17. **Beispiel-Use-Case**: - **Anfrage**: "Erstelle eine Social-Media-Kampagne für Kunde Acme Corp, Zielgruppe sind junge Erwachsene (18-25), Budget 5000€." - **Ablauf**: 1. Orchestrator lädt Kontext: Aktueller Nutzer ist Agent Smith, aktueller Kunde ist Acme Corp. 2. Erinnerungsagent: - Findet letzte Interaktion mit Acme Corp: "Letzte Kampagne war im März für Produkt X, Zielgruppe 30-40, Budget 10000€." - Findet Präferenzen: "Acme mag blaue Farben und humorvolle Tone." 3. Wissensagent: - Findet in Dokumenten: "Für Zielgruppe 18-25 funktionieren TikTok und Instagram am besten." - Findet Branchenstandards: "Budgetaufteilung: 60% Content, 30% Ads, 10% Analyse." 4. Kreativagent: - Generiert Kampagnenideen: "TikTok-Challenge mit Hashtag #AcmeFun, Influencer-Kooperationen." - Erstellt Textentwürfe: "Mach mit bei der #AcmeFun Challenge und gewinne coole Preise!" 5. Tool-Agent: - Erstellt Anzeigen in Google Ads und Meta Ads über die Adapter. - Plant Posts in einem Social-Media-Tool (z.B. Buffer). 6. Antwort an Nutzer: - "Hier ist dein Kampagnenplan für Acme Corp: - Zielgruppe: 18-25, Plattformen: TikTok (60%), Instagram (30%), Facebook (10%). - Budgetaufteilung: 3000€ Content, 1500€ Ads, 500€ Analyse. - Kreativ: TikTok-Challenge mit #AcmeFun, 3 Influencer-Posts. - Zeitplan: 4 Wochen, Start nächsten Montag. - Anzeigen wurden in Google Ads und Meta erstellt. Links: [Link1], [Link2]." 18. **Kostenoptimierung**: - **LLMs**: - Nutze kleine, lokale Modelle (z.B. Mistral-7B, Phi-2) für die meisten Aufgaben. - Nur für komplexe Aufgaben (z.B. kreative Texte) größere Modelle oder Cloud-APIs nutzen. - Caching: Speichere häufige LLM-Antworten in Redis (z.B. für Standardfragen). - **Datenbanken**: - Beginne mit SQLite oder FAISS auf der Festplatte (kostenlos). - Wechsle zu Cloud-Diensten nur, wenn Skalierung nötig ist. - **Tools**: - Nutze kostenlose APIs wo möglich (z.B. Google Ads hat ein kostenloses Testkonto). - Für interne Tools: Baue Mock-Adapter, die keine echten API-Aufrufe machen (für Entwicklung). 19. **Zeitplan (Realistisch)**: | Phase | Dauer | Ergebnis | |---------------------|----------|-----------------------------------| | Recherche & Planung | 1 Woche | Tech-Stack festgelegt, Dokumente gesammelt | | MVP (RAG + Erinnerung) | 2 Wochen | Funktionierendes RAG-System mit einfacher Erinnerung | | Tool-Integration | 2 Wochen | Erste Tool-Adapter (z.B. Google Ads) | | Test & Iteration | 2 Wochen | Feedback einarbeiten, Bugs fixen | | Skalierung | 4 Wochen | Mehr Agenten, bessere Erinnerung, Performance-Optimierung | | **Gesamt** | **11 Wochen** | Voll funktionierendes System | *Hinweis*: Mit einem kleinen Team (1-2 Entwickler) ist das machbar. Bei Alleinarbeit vielleicht etwas länger. 20. **Empfohlene Ressourcen**: - **Tutorials**: - [LangChain RAG Tutorial](https://python.langchain.com/docs/use_cases/question_answering/) - [Neo4j + Python](https://neo4j.com/docs/python-manual/current/) - [AutoGen Multi-Agenten](https://microsoft.github.io/autogen/docs/Use-Cases/agent_chat) - **Beispielprojekte**: - [AutoGen Agent Chat](https://github.com/microsoft/autogen/tree/main/samples) - [LangChain + Neo4j](https://github.com/langchain-ai/langchain/tree/master/libs/langchain/langchain/graphs) - **Bücher/Papers**: - [Agentic AI: Building Next-Gen Applications with LLMs](https://arxiv.org/abs/2308.11454) (Paper über Agenten-Architekturen) - [Designing Agent Systems with LLMs](https://lilianweng.github.io/posts/2023-06-23-agent/) (Blogpost von Lil'Log) 21. **Mögliche Fallstricke & Lösungen**: | Problem | Lösung | |----------------------------------|------------------------------------------------------------------------| | RAG gibt irrelevante Ergebnisse | Nutze bessere Embeddings (z.B. `bge-small-en-v1.5`) oder Hybrid-Suche (Keyword + Vektor). | | Erinnerung ist unvollständig | Implementiere ein Event-Sourcing-System, das alle Interaktionen speichert. | | Tools sind langsam | Asynchrone Verarbeitung (z.B. Celery) für Tool-Aufrufe. | | LLMs sind zu teuer | Nutze lokale Modelle oder Caching. | | Kontext geht verloren | Nutze Session-IDs und speichere Kontext in Redis. | | Agenten kooperieren nicht gut | Nutze ein Framework wie AutoGen oder CrewAI, das Koordination unterstützt. | 22. **Zusammenfassung der Strategie**: - **Schritt 1: MVP mit RAG + einfache Erinnerung** - Technologien: FAISS, Sentence-Transformers, Python (FastAPI/Streamlit), JSON für Erinnerung. - Ziel: Grundlegende Wissensabfrage + einfache Erinnerungsfunktion. - **Schritt 2: Graphbasierte Erinnerung & Kontext** - Technologien: Neo4j, Redis. - Ziel: Dynamische Erinnerung über Sessions hinweg + Kontextmanagement. - **Schritt 3: Tool-Integration** - Technologien: Eigenbau-Adapter, FastAPI für Tool-Schnittstelle. - Ziel: Agenten können externe Tools nutzen (z.B. Werbeplattformen). - **Schritt 4: Skalierung & Optimierung** - Technologien: Caching (Redis), asynchrone Verarbeitung (Celery), Monitoring (Prometheus). - Ziel: System stabilisieren und für Produktion vorbereiten. - **Kosten**: ~0-50€/Monat (je nach Cloud-Nutzung). - **Zeit**: ~11 Wochen für ein voll funktionierendes System (mit 1-2 Entwicklern). 23. **Finaler Vorschlag**: Hier ist ein konkreter, umsetzbarer Plan für dich: --- ### **Strategie für dein Agentensystem** #### **1. Ziele definieren** - Was soll das System können? - Beispiel: "Agenten sollen Werbekampagnen erstellen, Kundenhistorie kennen, und Tools wie Google Ads nutzen." - Welche Daten hast du? - Beispiel: Kundenakten, vergangene Kampagnen, Branchenreports. #### **2. Tech-Stack auswählen (kostengünstig & schnell)** | Komponente | Empfohlene Technologie | Alternative | Kosten | |---------------------|------------------------|----------------------|---------| | Agenten-Framework | CrewAI | AutoGen, LangGraph | Kostenlos | | Wissensdatenbank | FAISS + Sentence-Transformers | Weaviate, Pinecone | Kostenlos* | | Erinnerungsdatenbank| Neo4j (Community) | Dgraph, PostgreSQL | Kostenlos | | Kontextspeicher | Redis | In-Memory (Python) | Kostenlos | | LLMs | Ollama + Mistral-7B | Hugging Face, OpenAI | Kostenlos* | | Backend | FastAPI | Flask | Kostenlos | | Frontend | Streamlit | Gradio, React | Kostenlos | | Tool-Adapter | Eigenbau (Python) | - | Kostenlos | *Weaviate/Pinecone haben kostenlose Tiers, Ollama ist lokal kostenlos (aber Hardwarekosten). #### **3. Schritt-für-Schritt-Umsetzung** **📌 Phase 1: Grundlagen (1-2 Wochen)** - **Ziel**: Einfaches RAG-System + Basisfrontend. - **Schritte**: 1. **Daten sammeln**: - Sammle 10-20 Dokumente (PDFs, Word, Texte) mit relevantem Wissen (z.B. Werbestrategien, Kundeninfos). - Speichere sie in einem Ordner `data/`. 2. **RAG-System aufbauen**: - Installiere: `pip install langchain faiss-cpu sentence-transformers` - Beispielcode für RAG: ```python from langchain.document_loaders import DirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain.llms import Ollama # 1. Dokumente laden loader = DirectoryLoader('./data/', glob="**/*.pdf") documents = loader.load() # 2. Texte splitten text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) texts = text_splitter.split_documents(documents) # 3. Embeddings erstellen embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") # 4. Vektordatenbank erstellen db = FAISS.from_documents(texts, embeddings) # 5. RAG-Kette erstellen llm = Ollama(model="mistral") qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=db.as_retriever()) response = qa.run("Was sind die besten Werbestrategien für junge Zielgruppen?") print(response) ``` 3. **Frontend bauen**: - Installiere Streamlit: `pip install streamlit` - Erstelle `app.py`: ```python import streamlit as st from langchain_community.vectorstores import FAISS from langchain.embeddings import HuggingFaceEmbeddings from langchain.chains import RetrievalQA from langchain.llms import Ollama # Lade RAG-System (wie oben) # ... st.title("Werbeagentur Agent") query = st.text_input("Frage stellen:") if query: response = qa.run(query) st.write("Antwort:", response) ``` - Starte mit: `streamlit run app.py` 4. **Erinnerung hinzufügen (einfach)**: - Nutze eine JSON-Datei für einfache Erinnerung: ```python import json def save_memory(user_id, memory): with open("memory.json", "r+") as f: data = json.load(f) if user_id not in data: data[user_id] = [] data[user_id].append({"time": str(datetime.now()), "memory": memory}) f.seek(0) json.dump(data, f) def get_memory(user_id): with open("memory.json", "r") as f: data = json.load(f) return data.get(user_id, []) ``` - Integriere das in deine RAG-Kette (z.B. füge Erinnerung als Kontext zur Query hinzu). **📌 Phase 2: Erinnerung & Kontext (2-3 Wochen)** - **Ziel**: Dynamische Erinnerung mit Neo4j + Kontextmanagement. - **Schritte**: 1. **Neo4j einrichten**: - Lade [Neo4j Desktop](https://neo4j.com/download/) herunter und installiere es. - Erstelle eine neue Datenbank (z.B. "AdAgencyMemory"). - Starte die Datenbank und notiere die URL (z.B. `bolt://localhost:7687`) und Credentials. 2. **Neo4j mit Python verbinden**: - Installiere: `pip install neo4j` - Beispielcode: ```python from neo4j import GraphDatabase class MemoryAgent: def __init__(self, uri, user, password): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def close(self): self.driver.close() def add_interaction(self, user_id, agent_id, message): with self.driver.session() as session: session.write_transaction( self._add_interaction, user_id, agent_id, message ) @staticmethod def _add_interaction(tx, user_id, agent_id, message): tx.run(""" MERGE (u:User {id: $user_id}) MERGE (a:Agent {id: $agent_id}) CREATE (u)-[r:INTERACTED {time: datetime(), message: $message}]->(a) """, user_id=user_id, agent_id=agent_id, message=message) def get_last_interaction(self, user_id): with self.driver.session() as session: result = session.read_transaction( self._get_last_interaction, user_id ) return result @staticmethod def _get_last_interaction(tx, user_id): result = tx.run(""" MATCH (u:User {id: $user_id})-[r:INTERACTED]->(a:Agent) RETURN r.message AS message, r.time AS time ORDER BY r.time DESC LIMIT 1 """, user_id=user_id) return result.single() ``` 3. **Kontextmanagement mit Redis**: - Installiere Redis lokal oder nutze einen Cloud-Dienst (z.B. [Redis Cloud Free Tier](https://redis.com/try-free/)). - Installiere: `pip install redis` - Beispielcode: ```python import redis import json class ContextManager: def __init__(self, host='localhost', port=6379, db=0): self.r = redis.Redis(host=host, port=port, db=db) def set_context(self, session_id, context): self.r.set(session_id, json.dumps(context)) def get_context(self, session_id): data = self.r.get(session_id) return json.loads(data) if data else {} ``` 4. **Agenten integrieren**: - Erstelle einen `OrchestratorAgent`, der: - Den Kontext aus Redis lädt. - Die Erinnerung aus Neo4j lädt. - Die RAG-Kette nutzt. - Die Antwort generiert und den Kontext/Erinnerung aktualisiert. - Beispiel: ```python class OrchestratorAgent: def __init__(self): self.qa = ... # RAG-Kette aus Phase 1 self.memory = MemoryAgent("bolt://localhost:7687", "neo4j", "password") self.context = ContextManager() def handle_query(self, user_id, session_id, query): # 1. Kontext laden context = self.context.get_context(session_id) if not context: context = {"user_id": user_id, "session_id": session_id} # 2. Erinnerung laden last_interaction = self.memory.get_last_interaction(user_id) if last_interaction: context["last_interaction"] = last_interaction["message"] # 3. RAG mit Kontext augmented_query = f"{query}\nKontext: {context}" response = self.qa.run(augmented_query) # 4. Erinnerung speichern self.memory.add_interaction(user_id, "orchestrator", query) # 5. Kontext aktualisieren context["last_query"] = query self.context.set_context(session_id, context) return response ``` **📌 Phase 3: Tool-Integration (2-4 Wochen)** - **Ziel**: Agenten können externe Tools nutzen (z.B. Google Ads, Canva). - **Schritte**: 1. **Tool-Schnittstelle definieren**: - Erstelle eine Basis-Klasse für Tool-Adapter: ```python from abc import ABC, abstractmethod class ToolAdapter(ABC): @abstractmethod def execute(self, action: str, params: dict) -> dict: pass @abstractmethod def get_name(self) -> str: pass ``` 2. **Beispiel-Adapter für Google Ads**: - Installiere Google Ads API Client: `pip install google-ads-api` - Beispielcode: ```python from google.ads.googleads.client import GoogleAdsClient from google.ads.googleads.errors import GoogleAdsException class GoogleAdsAdapter(ToolAdapter): def __init__(self, developer_token, client_id, client_secret, refresh_token): self.client = GoogleAdsClient.load_from_dict({ "developer_token": developer_token, "client_id": client_id, "client_secret": client_secret, "refresh_token": refresh_token, }) def get_name(self): return "google_ads" def execute(self, action, params): if action == "create_campaign": return self._create_campaign(params) elif action == "get_campaigns": return self._get_campaigns(params) else: raise ValueError(f"Unknown action: {action}") def _create_campaign(self, params): # Implementierung der Google Ads API Aufrufe # Beispiel: https://developers.google.com/google-ads/api/docs/start customer_id = params.get("customer_id") campaign_name = params.get("name") # ... (komplexe API-Logik) return {"status": "success", "campaign_id": "12345"} def _get_campaigns(self, params): # ... pass ``` 3. **Tool-Agent erstellen**: - Der Tool-Agent verwaltet alle Adapter und führt Aktionen aus: ```python class ToolAgent: def __init__(self): self.adapters = { "google_ads": GoogleAdsAdapter(...), # Weitere Adapter hier } def execute(self, tool_name, action, params): adapter = self.adapters.get(tool_name) if not adapter: raise ValueError(f"Tool {tool_name} not found") return adapter.execute(action, params) ``` 4. **Orchestrator erweitern**: - Integriere den Tool-Agent in den Orchestrator: ```python class OrchestratorAgent: def __init__(self): # ... (vorheriger Code) self.tool_agent = ToolAgent() def handle_query(self, user_id, session_id, query): # ... (vorheriger Code) # 5. Tools nutzen (falls nötig) if "erstelle kampagne" in query.lower(): # Extrahiere Parameter aus der Query (z.B. mit LLM oder Regex) params = self._extract_params(query) result = self.tool_agent.execute("google_ads", "create_campaign", params) response += f"\nKampagne erstellt: {result}" return response def _extract_params(self, query): # Einfache Extraktion (später mit LLM verbessern) params = {} if "budget" in query: # Extrahiere Budget (z.B. mit Regex) import re match = re.search(r"budget (\d+)", query) if match: params["budget"] = int(match.group(1)) # ... return params ``` 5. **Frontend erweitern**: - Füge eine Option hinzu, um Tools auszuwählen: ```python st.title("Werbeagentur Agent") query = st.text_input("Frage stellen:") tool = st.selectbox("Tool auswählen:", ["kein Tool", "google_ads", "canva"]) tool_action = st.text_input("Tool-Aktion (z.B. create_campaign):") tool_params = st.text_area("Tool-Parameter (JSON):") if st.button("Senden"): if tool != "kein Tool" and tool_action: params = json.loads(tool_params) if tool_params else {} result = tool_agent.execute(tool, tool_action, params) st.write("Tool-Ergebnis:", result) else: response = orchestrator.handle_query("user123", "session456", query) st.write("Antwort:", response) ``` **📌 Phase 4: Skalierung & Optimierung (laufend)** - **Ziel**: System stabilisieren, Performance verbessern, mehr Funktionen hinzufügen. - **Schritte**: 1. **Caching**: - Nutze Redis für Caching von RAG-Ergebnissen: ```python from langchain.cache import RedisCache # In deiner RAG-Kette: cache = RedisCache(redis_=redis_client) qa = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=db.as_retriever(), cache=cache ) ``` 2. **Asynchrone Verarbeitung**: - Nutze Celery für lange Laufzeiten (z.B. Tool-Aufrufe): ```python from celery import Celery app = Celery('tasks', broker='redis://localhost:6379/0') @app.task def long_running_task(query): # ... return response # Aufrufen: result = long_running_task.delay(query) ``` 3. **Monitoring**: - Nutze Prometheus + Grafana für Metriken (z.B. Antwortzeiten, Fehlerraten). 4. **Mehr Agenten hinzufügen**: - Spezialisierte Agenten für: - Kreativität (Texterstellung) - Analyse (Kampagnenperformance) - Planung (Zeitpläne, Budgets) - Beispiel: ```python class CreativeAgent: def __init__(self): self.llm = Ollama(model="mistral") def generate_ideas(self, prompt, context): augmented_prompt = f""" Du bist ein kreativer Werbetexter. Kontext: {context} Aufgabe: {prompt} Generiere 3 kreative Ideen für eine Werbekampagne. """ return self.llm(augmented_prompt) # Im Orchestrator: def handle_query(self, ...): # ... if "ideen" in query.lower(): ideas = self.creative_agent.generate_ideas(query, context) response += f"\nIdeen: {ideas}" ``` 5. **Daten aktualisieren**: - Erstelle einen Prozess, um regelmäßig neue Dokumente in die Vektordatenbank zu laden. - Beispiel: ```python def update_knowledge_base(new_documents): texts = text_splitter.split_documents(new_documents) db.add_documents(texts) db.save_local("faiss_index") ``` **📌 Phase 5: Deployment (1 Woche)** - **Option 1: Lokale Bereitstellung (für Entwicklung/Tests)** - Nutze Docker, um alle Komponenten in Containern zu betreiben: ```dockerfile # Dockerfile für Backend FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] # docker-compose.yml version: '3' services: backend: build: . ports: - "8000:8000" depends_on: - redis - neo4j redis: image: redis:alpine ports: - "6379:6379" neo4j: image: neo4j:community ports: - "7474:7474" - "7687:7687" environment: NEO4J_AUTH: neo4j/password ``` - Starte mit: `docker-compose up -d` - **Option 2: Cloud-Bereitstellung (für Produktion)** - Nutze günstige Cloud-Anbieter: - **Hetzner Cloud**: ~5€/Monat für einen Server mit 4GB RAM. - **Oracle Cloud Free Tier**: Kostenlos (aber komplexer). - Deployment-Schritte: 1. Server erstellen (z.B. Ubuntu 22.04). 2. Docker und Docker Compose installieren. 3. Code auf den Server kopieren (z.B. mit `scp` oder Git). 4. `docker-compose up -d` ausführen. 5. Domain und HTTPS einrichten (z.B. mit Nginx + Let's Encrypt). #### **4. Kostenübersicht** | Komponente | Kosten (Monatlich) | Anmerkungen | |---------------------|--------------------|----------------------------------| | Server (Hetzner) | ~5-20€ | Abhängig von der Größe | | LLMs (lokal) | 0€ | Ollama + Mistral-7B | | Neo4j | 0€ | Community Edition (self-hosted) | | Redis | 0€ | Self-hosted oder kostenloser Cloud-Tier | | Google Ads API | 0€ | Kostenlos für Entwicklerkonten | | **Gesamt** | **5-20€** | Je nach Server-Größe | *Hinweis*: Falls du Cloud-LLMs (z.B. OpenAI) nutzt, kommen zusätzliche Kosten hinzu (z.B. ~0.01-0.10€ pro Anfrage). #### **5. Zeitplan (Realistisch)** | Phase | Dauer | Aufwand (Personentage) | Ergebnis | |---------------------|----------|------------------------|-----------------------------------| | Phase 1: MVP | 2 Wochen | 10 | RAG + einfaches Frontend | | Phase 2: Erinnerung | 3 Wochen | 15 | Neo4j + Kontextmanagement | | Phase 3: Tools | 3 Wochen | 15 | Tool-Adapter + Orchestrator | | Phase 4: Skalierung | 2 Wochen | 10 | Caching, Monitoring, mehr Agenten | | Phase 5: Deployment | 1 Woche | 5 | Produktivsystem | | **Gesamt** | **11 Wochen** | **55 Personentage** | Voll funktionierendes System | *Hinweis*: Mit einem Entwickler (Vollzeit) sind das ~11 Wochen. Mit 2 Entwicklern könnte es in ~6 Wochen geschafft werden. #### **6. Empfohlene erste Schritte** 1. **Diese Woche**: - Dokumente sammeln (10-20 relevante Dateien). - Lokale Umgebung einrichten: - Python 3.10+ installieren. - Docker installieren. - Ollama installieren und Mistral-7B herunterladen (`ollama pull mistral`). - Einfaches RAG-System mit FAISS und Streamlit bauen (siehe Phase 1). 2. **Nächste Woche**: - Neo4j lokal installieren und erste Daten laden. - Erinnerungsagent implementieren (Phase 2). 3. **Übernächste Woche**: - Ersten Tool-Adapter (z.B. für Google Ads) implementieren (Phase 3). #### **7. Tools & Bibliotheken im Überblick** | Zweck | Empfohlene Bibliothek | Alternative | Link | |---------------------|------------------------|----------------------|-------------------------------| | Agenten-Framework | CrewAI | AutoGen, LangGraph | [CrewAI](https://github.com/joaomdmoura/crewAI) | | RAG | LangChain + FAISS | LlamaIndex, Weaviate | [LangChain](https://python.langchain.com/) | | Graphdatenbank | Neo4j | Dgraph, PostgreSQL | [Neo4j](https://neo4j.com/) | | Kontextspeicher | Redis | SQLite, In-Memory | [Redis](https://redis.io/) | | LLMs (lokal) | Ollama + Mistral-7B | Hugging Face | [Ollama](https://ollama.ai/) | | Backend | FastAPI | Flask | [FastAPI](https://fastapi.tiangolo.com/) | | Frontend | Streamlit | Gradio, React | [Streamlit](https://streamlit.io/) | | Tool-Adapter | Eigenbau | - | - | | Workflow-Orchestrierung | Celery | Airflow, Prefect | [Celery](https://www.celeryq.dev/) | #### **8. Beispiel-Architektur-Diagramm** ``` +-------------------+ +-------------------+ +-------------------+ | | | | | | | Nutzer |------>| Frontend |------>| Orchestrator | | (Browser) | | (Streamlit) | | Agent | +-------------------+ +-------------------+ +--------+----------+ | +-------------------+ +-------------------+ | | | | | | | Tool-Adapter |<------| Tool-Agent |<------------+ | (Google Ads) | | | +-------------------+ +-------------------+ | +-------------------+ +--------+----------+ +-------------------+ | | | | | | | RAG (FAISS) |<------| Wissensagent | | LLMs (Ollama) | | | | | | | +-------------------+ +--------+----------+ +-------------------+ | +-------------------+ +--------+----------+ | | | | | Neo4j |<------| Erinnerungs- | | (GraphDB) | | agent | +-------------------+ +-------------------+ | +-------------------+ | | | | | Redis |<--------------+ | (Kontext) | +-------------------+ ``` #### **9. Mögliche Erweiterungen für die Zukunft** - **Multi-Modalität**: - Füge Bild- und Videoverarbeitung hinzu (z.B. mit [CLIP](https://openai.com/research/clip) für Bilderkennung). - Nutze [Stable Diffusion](https://stability.ai/) für Bildgenerierung (z.B. für Anzeigenbilder). - **Echtzeit-Daten**: - Integriere Live-Datenquellen (z.B. Google Analytics API für Kampagnenperformance). - Nutze Webhooks für Echtzeit-Updates (z.B. wenn eine Kampagne gestartet wird). - **Benutzerfeedback**: - Füge ein Feedback-System hinzu (z.B. "War diese Antwort hilfreich?"). - Nutze das Feedback, um das System zu verbessern (z.B. mit [RLHF](https://huggingface.co/blog/rlhf)). - **Sicherheit**: - Implementiere Authentifizierung (z.B. mit [FastAPI Users](https://fastapi-users.github.io/fastapi-users/)). - Nutze API-Keys für Tool-Zugriff. - **Skalierbarkeit**: - Nutze Kubernetes für Container-Orchestrierung (falls viele Nutzer). - Implementiere Load Balancing für das Backend. #### **10. Zusammenfassung: Dein Aktionsplan** | Schritt | Aufgabe | Zeitaufwand | Tools/Bibliotheken | |---------|---------|-------------|--------------------| | 1 | Dokumente sammeln | 1 Tag | - | | 2 | Lokale Umgebung einrichten (Python, Docker, Ollama) | 1 Tag | Python, Docker, Ollama | | 3 | Einfaches RAG-System bauen | 3 Tage | LangChain, FAISS, Sentence-Transformers | | 4 | Streamlit-Frontend erstellen | 1 Tag | Streamlit | | 5 | Einfache Erinnerung (JSON) hinzufügen | 2 Tage | Python (json) | | 6 | Neo4j einrichten und Erinnerungsagent implementieren | 5 Tage | Neo4j, neo4j Python Driver | | 7 | Kontextmanagement mit Redis hinzufügen | 2 Tage | Redis, redis-py | | 8 | Tool-Adapter für 1-2 Tools implementieren | 5 Tage | Google Ads API, etc. | | 9 | Orchestrator-Agent erstellen | 3 Tage | Python | | 10 | Frontend erweitern (Tool-Integration) | 2 Tage | Streamlit | | 11 | Caching und Performance-Optimierungen | 3 Tage | Redis, Celery | | 12 | Monitoring einrichten | 2 Tage | Prometheus, Grafana | | 13 | Deployment (lokal oder Cloud) | 3 Tage | Docker, Hetzner Cloud | | **Gesamt** | | **~35 Tage** | | *Hinweis*: Das ist eine grobe Schätzung. Je nach Erfahrung kann es schneller oder langsamer gehen. #### **11. Tipps für den Erfolg** - **Starte klein**: Beginne mit einem MVP und baue schrittweise aus. - **Nutze Open Source**: Die meisten benötigten Tools sind kostenlos und Open Source. - **Dokumentiere alles**: Besonders die Tool-Adapter und Datenbank-Schemas. - **Teste regelmäßig**: Schreibe Unit-Tests für deine Agenten (z.B. mit `pytest`). - **Nutze Versionierung**: Git für Code, und z.B. [DVC](https://dvc.org/) für Daten. - **Community nutzen**: Es gibt viele Open-Source-Projekte und Foren (z.B. [LangChain Discord](https://discord.gg/80R5K7HxWV)), die helfen können. #### **12. Beispiel-Code-Repository-Struktur** ``` /ad_agency_agent ├── /data # Dokumente für RAG │ ├── kampagnen.pdf │ ├── kundeninfos.docx │ └── ... ├── /docs # Dokumentation │ ├── architecture.md │ └── api.md ├── /src │ ├── /agents # Agenten-Klassen │ │ ├── orchestrator.py │ │ ├── knowledge_agent.py │ │ ├── memory_agent.py │ │ ├── creative_agent.py │ │ └── tool_agent.py │ ├── /adapters # Tool-Adapter │ │ ├── google_ads_adapter.py │ │ ├── canva_adapter.py │ │ └── base_adapter.py │ ├── /utils # Hilfsfunktionen │ │ ├── rag.py │ │ ├── neo4j_utils.py │ │ └── redis_utils.py │ ├── main.py # FastAPI/Streamlit App │ └── config.py # Konfiguration (API-Keys, etc.) ├── /tests # Tests │ ├── test_agents.py │ └── test_adapters.py ├── requirements.txt # Python Abhängigkeiten ├── docker-compose.yml # Docker Konfiguration └── README.md # Projektbeschreibung ``` #### **13. Empfohlene Lernressourcen** - **RAG**: - [LangChain RAG Tutorial](https://python.langchain.com/docs/use_cases/question_answering/) - [FAISS GitHub](https://github.com/facebookresearch/faiss) - **Neo4j**: - [Neo4j Python Driver Docs](https://neo4j.com/docs/python-manual/current/) - [Neo4j GraphAcademy](https://neo4j.com/graphacademy/) - **Agenten**: - [AutoGen Docs](https://microsoft.github.io/autogen/docs/) - [CrewAI Docs](https://docs.crewai.com/) - [LangGraph Docs](https://langchain-ai.github.io/langgraph/) - **LLMs**: - [Ollama GitHub](https://github.com/jmorganca/ollama) - [Hugging Face Transformers](https://huggingface.co/docs/transformers/index) - **Deployment**: - [Docker Docs](https://docs.docker.com/) - [FastAPI Deployment](https://fastapi.tiangolo.com/deployment/) #### **14. Mögliche Herausforderungen & Lösungen** | Herausforderung | Lösung | |-----------------|--------| | RAG gibt schlechte Ergebnisse | Nutze bessere Embeddings (z.B. `bge-base-en-v1.5`) oder Hybrid-Suche (Keyword + Vektor). | | Neo4j ist langsam | Optimierte Abfragen schreiben, Indizes erstellen. | | LLMs sind zu langsam | Nutze kleinere Modelle (z.B. `mistral-7b-instruct`) oder Quantisierung (z.B. `mistral-7b-instruct.Q4_K_M.gguf`). | | Tool-Adapter sind komplex | Beginne mit einfachen Tools (z.B. eine Mock-API für Tests). | | Erinnerung ist unvollständig | Speichere alle Interaktionen als Events in Neo4j. | | Kontext geht verloren | Nutze Session-IDs und speichere Kontext in Redis. | | Skalierbarkeit | Nutze asynchrone Verarbeitung (Celery) und Caching (Redis). | #### **15. Finaler Rat** - **Fang klein an und iteriere**: Baue zuerst ein einfaches System, das eine konkrete Aufgabe löst (z.B. "Antworte auf Fragen zu vergangenen Kampagnen"). Dann füge nach und nach mehr Funktionen hinzu. - **Nutze bestehende Frameworks**: CrewAI, LangChain und AutoGen sparen dir viel Zeit. - **Dokumentiere deine Daten**: Besonders wichtig für RAG und Erinnerung. Ohne gute Daten funktioniert das System nicht. - **Teste mit echten Use Cases**: Frage deine Kollegen, welche Funktionen sie brauchen, und baue diese zuerst. - **Kosten im Auge behalten**: Nutze lokale Modelle und kostenlose Cloud-Tiers, wo immer möglich. --- ### **Zusammenfassung der Strategie** 1. **Starte mit einem MVP** (RAG + einfaches Frontend) in 1-2 Wochen. 2. **Füge Erinnerung & Kontext hinzu** (Neo4j + Redis) in 2-3 Wochen. 3. **Integriere Tools** (Adapter für Google Ads, etc.) in 2-4 Wochen. 4. **Skalieren & Optimieren** (Caching, Monitoring, mehr Agenten) in 2-4 Wochen. 5. **Deploye das System** (Docker + Cloud) in 1 Woche. **Gesamtkosten**: ~5-20€/Monat (je nach Server-Größe). **Gesamtzeit**: ~11 Wochen (mit einem Entwickler). **Tech-Stack Empfehlung**: - Agenten: CrewAI - RAG: LangChain + FAISS - Erinnerung: Neo4j - Kontext: Redis - LLMs: Ollama + Mistral-7B - Backend: FastAPI - Frontend: Streamlit - Tools: Eigenbau-Adapter **Erste Schritte diese Woche**: 1. Dokumente sammeln. 2. Lokale Umgebung einrichten (Python, Docker, Ollama). 3. Einfaches RAG-System mit FAISS und Streamlit bauen. Hier ist eine **komplette, praxisnahe Strategie**, um dein **tool-unabhängiges Agentensystem mit komplexem Wissen/Erinnerung/Kontext** für deine Werbeagentur **schnell (11 Wochen) und kostengünstig (5–20€/Monat)** aufzubauen. Die Lösung geht **über RAG und einfache Knowledge Graphs hinaus** und kombiniert **dynamische Erinnerung, Kontextmanagement und Multi-Agenten-Koordination** – alles mit Open-Source-Tools. --- --- --- ## 🎯 **Zieldefinition: Was soll dein System können?** Dein System sollte folgende **Kernfähigkeiten** haben (priorisiert für Werbeagenturen): | **Fähigkeit** | **Beispiel** | **Technische Umsetzung** | |-----------------------------|-----------------------------------------------------------------------------|---------------------------------------------| | **Wissen (RAG+)** | "Welche Strategien funktionieren für Zielgruppe 18–25?" | Vektordatenbank (FAISS) + GraphDB (Neo4j) | | **Erinnerung** | "Was war die letzte Kampagne für Kunde X?" | Neo4j (Zeitstempel + Beziehungen) | | **Kontext** | "Wir sind in Phase 2 der Kampagne für Kunde Y." | Redis (Session-Speicher) | | **Tool-Nutzung** | "Erstelle eine Google Ads Kampagne mit Budget 5000€." | Adapter-Pattern (Python-Klassen) | | **Multi-Agenten-Koordination** | "Agent A plant die Kampagne, Agent B erstellt die Texte." | CrewAI/AutoGen | | **Langzeitgedächtnis** | "Kunde X mag blaue Farben und hat ein Budget von 10.000€." | Neo4j + Event Sourcing | --- --- --- ## 🚀 **Strategie: 5 Phasen in 11 Wochen (Kosten: 5–20€/Monat)** ### **📌 Phase 1: MVP mit RAG + einfache Erinnerung (2 Wochen)** **Ziel**: Grundlegendes System, das auf Dokumente antwortet und einfache Erinnerungen speichert. #### **🔧 Tech-Stack** | Komponente | Technologie | Kosten | Aufwand | |---------------------|---------------------------|---------|---------| | **Agenten-Framework** | CrewAI | 0€ | Niedrig | | **RAG** | FAISS + Sentence-Transformers | 0€ | Niedrig | | **Erinnerung** | JSON-Datei | 0€ | Niedrig | | **LLMs** | Ollama + Mistral-7B | 0€* | Niedrig | | **Frontend** | Streamlit | 0€ | Niedrig | | **Backend** | FastAPI | 0€ | Niedrig | \* *Ollama ist lokal kostenlos, braucht aber eine GPU für gute Performance.* --- #### **🛠 Schritt-für-Schritt-Anleitung** **📌 Schritt 1: Umgebung einrichten (1 Tag)** 1. **Python 3.10+** installieren. 2. **Docker** installieren (für Neo4j/Redis später). 3. **Ollama** installieren und Mistral-7B herunterladen: ```bash curl -fsSL https://ollama.ai/install.sh | sh ollama pull mistral ``` 4. **Projektverzeichnis** erstellen: ```bash mkdir ad_agent && cd ad_agent python -m venv venv && source venv/bin/activate # (Linux/Mac) oder venv\Scripts\activate (Windows) pip install langchain faiss-cpu sentence-transformers streamlit fastapi uvicorn ``` **📌 Schritt 2: Dokumente sammeln (1 Tag)** - Sammle **10–20 Dokumente** (PDFs, Word, Texte) mit: - Werbestrategien - Kundenbriefings - Branchenreports - vergangenen Kampagnen - Speichere sie in `/data/`. **📌 Schritt 3: RAG-System aufbauen (3 Tage)** Erstelle `src/rag.py`: ```python from langchain.document_loaders import DirectoryLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import HuggingFaceEmbeddings from langchain.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain.llms import Ollama # 1. Dokumente laden loader = DirectoryLoader('./data/', glob="**/*.{pdf,docx,txt}") documents = loader.load() # 2. Texte in Chunks splitten text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len ) texts = text_splitter.split_documents(documents) # 3. Embeddings erstellen (kleines, schnelles Modell) embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2", model_kwargs={'device': 'cpu'} # oder 'cuda' für GPU ) # 4. Vektordatenbank erstellen db = FAISS.from_documents(texts, embeddings) db.save_local("faiss_index") # 5. RAG-Kette erstellen llm = Ollama(model="mistral") qa = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=db.as_retriever(search_kwargs={"k": 3}), return_source_documents=True ) # 6. Funktion zum Abfragen def query_rag(question): result = qa({"query": question}) return result["result"] ``` **📌 Schritt 4: Einfache Erinnerung hinzufügen (2 Tage)** Erstelle `src/memory.py`: ```python import json from datetime import datetime import os MEMORY_FILE = "memory.json" def init_memory(): if not os.path.exists(MEMORY_FILE): with open(MEMORY_FILE, "w") as f: json.dump({}, f) def save_memory(user_id: str, message: str, agent: str = "user"): init_memory() with open(MEMORY_FILE, "r+") as f: data = json.load(f) if user_id not in data: data[user_id] = [] data[user_id].append({ "time": str(datetime.now()), "agent": agent, "message": message }) f.seek(0) json.dump(data, f, indent=2) def get_memory(user_id: str, limit: int = 5): init_memory() with open(MEMORY_FILE, "r") as f: data = json.load(f) return data.get(user_id, [])[-limit:] # Letzte 'limit' Einträge ``` **📌 Schritt 5: Streamlit-Frontend erstellen (1 Tag)** Erstelle `app.py`: ```python import streamlit as st from src.rag import query_rag from src.memory import save_memory, get_memory # UI st.title("🚀 Werbeagentur Agent (MVP)") user_id = st.text_input("Nutzer-ID (z.B. agent_1)", value="user_1") query = st.text_input("Frage stellen:") if query: # Erinnerung speichern save_memory(user_id, query, agent="user") # RAG abfragen with st.spinner("Denke nach..."): response = query_rag(query) save_memory(user_id, response, agent="agent") st.write("🤖 **Antwort:**", response) # Erinnerung anzeigen st.subheader("📜 Letzte Interaktionen:") memories = get_memory(user_id) for mem in reversed(memories): st.write(f"**{mem['agent']}** ({mem['time']}): {mem['message']}") ``` Starte das Frontend: ```bash streamlit run app.py ``` → **Ergebnis**: Ein funktionierendes RAG-System mit einfacher Erinnerung! --- --- ### **📌 Phase 2: Dynamische Erinnerung & Kontext (3 Wochen)** **Ziel**: Neo4j für komplexe Erinnerungen + Redis für Kontextmanagement. #### **🔧 Tech-Stack** | Komponente | Technologie | Kosten | |---------------------|---------------------------|---------| | **Erinnerung** | Neo4j (Community) | 0€ | | **Kontext** | Redis | 0€ | --- #### **🛠 Schritt-für-Schritt-Anleitung** **📌 Schritt 1: Neo4j einrichten (1 Tag)** 1. **Neo4j Desktop** herunterladen und installieren: [Download](https://neo4j.com/download/). 2. Neue Datenbank erstellen (z.B. "AdAgencyMemory"). 3. Datenbank starten und **URI** (`bolt://localhost:7687`), **Benutzername** (`neo4j`) und **Passwort** notieren. **📌 Schritt 2: Neo4j mit Python verbinden (2 Tage)** Erstelle `src/neo4j_memory.py`: ```python from neo4j import GraphDatabase from datetime import datetime from typing import List, Dict, Optional class Neo4jMemory: def __init__(self, uri: str, user: str, password: str): self.driver = GraphDatabase.driver(uri, auth=(user, password)) def close(self): self.driver.close() def _ensure_indexes(self): with self.driver.session() as session: session.run("CREATE INDEX IF NOT EXISTS FOR (u:User) ON (u.id)") session.run("CREATE INDEX IF NOT EXISTS FOR (a:Agent) ON (a.id)") session.run("CREATE INDEX IF NOT EXISTS FOR (c:Customer) ON (c.id)") session.run("CREATE INDEX IF NOT EXISTS FOR (p:Campaign) ON (p.id)") def add_interaction( self, user_id: str, agent_id: str, message: str, customer_id: Optional[str] = None, campaign_id: Optional[str] = None ): with self.driver.session() as session: session.write_transaction( self._add_interaction_tx, user_id, agent_id, message, customer_id, campaign_id ) @staticmethod def _add_interaction_tx( tx, user_id: str, agent_id: str, message: str, customer_id: str, campaign_id: str ): query = """ MERGE (u:User {id: $user_id}) MERGE (a:Agent {id: $agent_id}) """ params = {"user_id": user_id, "agent_id": agent_id} if customer_id: query += "MERGE (c:Customer {id: $customer_id}) " params["customer_id"] = customer_id if campaign_id: query += "MERGE (p:Campaign {id: $campaign_id}) " params["campaign_id"] = campaign_id query += """ CREATE (u)-[r:INTERACTED { time: datetime(), message: $message }]->(a) """ if customer_id: query += "CREATE (u)-[:WORKS_FOR]->(c) " if campaign_id: query += "CREATE (a)-[:MANAGES]->(p) " tx.run(query, **params, message=message) def get_last_interactions( self, user_id: str, limit: int = 5, customer_id: Optional[str] = None ) -> List[Dict]: with self.driver.session() as session: result = session.read_transaction( self._get_last_interactions_tx, user_id, limit, customer_id ) return result @staticmethod def _get_last_interactions_tx(tx, user_id: str, limit: int, customer_id: str): query = """ MATCH (u:User {id: $user_id})-[r:INTERACTED]->(a:Agent) """ params = {"user_id": user_id, "limit": limit} if customer_id: query += "MATCH (u)-[:WORKS_FOR]->(c:Customer {id: $customer_id}) " params["customer_id"] = customer_id query += """ RETURN r.time AS time, r.message AS message, a.id AS agent_id, u.id AS user_id ORDER BY r.time DESC LIMIT $limit """ result = tx.run(query, **params) return [dict(record) for record in result] def get_customer_info(self, customer_id: str) -> Dict: with self.driver.session() as session: result = session.read_transaction( self._get_customer_info_tx, customer_id ) return result @staticmethod def _get_customer_info_tx(tx, customer_id: str): query = """ MATCH (c:Customer {id: $customer_id}) OPTIONAL MATCH (c)-[:PREFERS]->(p:Preference) OPTIONAL MATCH (c)-[:HAS_CAMPAIGN]->(camp:Campaign) RETURN c.id AS id, c.name AS name, c.budget AS budget, c.target_group AS target_group, collect(DISTINCT p.value) AS preferences, collect(DISTINCT camp.id) AS campaigns """ result = tx.run(query, customer_id=customer_id) return dict(result.single()) if result.single() else {} # Beispielnutzung: # memory = Neo4jMemory("bolt://localhost:7687", "neo4j", "password") # memory._ensure_indexes() # memory.add_interaction("user_1", "agent_1", "Frage zu Kampagne X", customer_id="acme") # print(memory.get_last_interactions("user_1")) # print(memory.get_customer_info("acme")) ``` **📌 Schritt 3: Redis für Kontext einrichten (1 Tag)** 1. **Redis lokal installieren** (oder [Redis Cloud Free Tier](https://redis.com/try-free/) nutzen): ```bash # Linux/Mac: brew install redis redis-server # im Hintergrund starten # Windows: Redis über Docker starten docker run -p 6379:6379 --name my-redis redis ``` 2. Erstelle `src/context.py`: ```python import redis import json from typing import Dict, Any class ContextManager: def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0): self.r = redis.Redis(host=host, port=port, db=db) def set_context(self, session_id: str, context: Dict[str, Any]): self.r.set(session_id, json.dumps(context)) def get_context(self, session_id: str) -> Dict[str, Any]: data = self.r.get(session_id) return json.loads(data) if data else {} def update_context(self, session_id: str, updates: Dict[str, Any]): context = self.get_context(session_id) context.update(updates) self.set_context(session_id, context) # Beispielnutzung: # context = ContextManager() # context.set_context("session_1", {"current_customer": "acme", "phase": "planning"}) # print(context.get_context("session_1")) ``` **📌 Schritt 4: Orchestrator-Agent erstellen (3 Tage)** Erstelle `src/agents/orchestrator.py`: ```python from typing import Dict, Any, Optional from src.rag import query_rag from src.neo4j_memory import Neo4jMemory from src.context import ContextManager class OrchestratorAgent: def __init__(self): self.rag = query_rag # aus Phase 1 self.memory = Neo4jMemory( uri="bolt://localhost:7687", user="neo4j", password="password" # Ändere das in der Produktion! ) self.memory._ensure_indexes() self.context = ContextManager() def handle_query( self, user_id: str, session_id: str, query: str, customer_id: Optional[str] = None ) -> str: # 1. Kontext laden context = self.context.get_context(session_id) if not context: context = {"user_id": user_id, "session_id": session_id} if customer_id: context["current_customer"] = customer_id # 2. Erinnerung laden (letzte 3 Interaktionen) memories = self.memory.get_last_interactions(user_id, limit=3) memory_context = "\n".join( [f"[{m['time']}] {m['agent_id']}: {m['message']}" for m in memories] ) # 3. Kundeninfo laden (falls vorhanden) customer_info = {} if customer_id: customer_info = self.memory.get_customer_info(customer_id) context["current_customer"] = customer_id # 4. Kontext für RAG vorbereiten augmented_query = f""" Frage: {query} Kontext: - Nutzer: {user_id} - Session: {session_id} - Aktueller Kunde: {context.get('current_customer', 'keiner')} - Letzte Interaktionen: {memory_context} - Kundeninfo: {customer_info} """ # 5. RAG abfragen response = self.rag(augmented_query) # 6. Erinnerung speichern self.memory.add_interaction( user_id=user_id, agent_id="orchestrator", message=query, customer_id=customer_id ) # 7. Kontext aktualisieren self.context.update_context(session_id, { "last_query": query, "last_response": response }) return response # Beispielnutzung: # orchestrator = OrchestratorAgent() # response = orchestrator.handle_query( # user_id="user_1", # session_id="session_1", # query="Erstelle eine Kampagne für Kunde Acme.", # customer_id="acme" # ) ``` **📌 Schritt 5: Streamlit-Frontend aktualisieren (1 Tag)** Aktualisiere `app.py`: ```python import streamlit as st from src.agents.orchestrator import OrchestratorAgent from src.neo4j_memory import Neo4jMemory from src.context import ContextManager # Initialisiere Agenten orchestrator = OrchestratorAgent() memory = Neo4jMemory("bolt://localhost:7687", "neo4j", "password") context = ContextManager() # UI st.title("🚀 Werbeagentur Agent (Phase 2)") user_id = st.text_input("Nutzer-ID", value="user_1") session_id = st.text_input("Session-ID", value="session_1") customer_id = st.text_input("Kunden-ID (optional)", value="acme") query = st.text_input("Frage stellen:") if query: with st.spinner("Denke nach..."): response = orchestrator.handle_query( user_id=user_id, session_id=session_id, query=query, customer_id=customer_id if customer_id else None ) st.write("🤖 **Antwort:**", response) # Erinnerung anzeigen st.subheader("📜 Letzte Interaktionen:") memories = memory.get_last_interactions(user_id, limit=5) for mem in memories: st.write(f"**{mem['agent_id']}** ({mem['time']}): {mem['message']}") # Kundeninfo anzeigen if customer_id: st.subheader("👤 Kundeninfo:") customer_info = memory.get_customer_info(customer_id) st.json(customer_info) ``` → **Ergebnis**: Ein System mit **dynamischer Erinnerung und Kontext**! --- --- ### **📌 Phase 3: Tool-Integration (3 Wochen)** **Ziel**: Agenten können externe Tools nutzen (z.B. Google Ads, Canva, Mailchimp). #### **🔧 Tech-Stack** | Komponente | Technologie | Kosten | |---------------------|---------------------------|---------| | **Tool-Adapter** | Eigenbau (Python) | 0€ | | **Google Ads API** | google-ads-api | 0€* | \* *Google Ads API ist kostenlos für Entwicklerkonten.* --- #### **🛠 Schritt-für-Schritt-Anleitung** **📌 Schritt 1: Basis-Adapter erstellen (1 Tag)** Erstelle `src/adapters/base_adapter.py`: ```python from abc import ABC, abstractmethod from typing import Dict, Any class ToolAdapter(ABC): """Basis-Klasse für alle Tool-Adapter.""" @abstractmethod def get_name(self) -> str: """Gibt den Namen des Tools zurück (z.B. 'google_ads').""" pass @abstractmethod def get_actions(self) -> list: """Gibt eine Liste der unterstützten Aktionen zurück (z.B. ['create_campaign', 'get_campaigns']).""" pass @abstractmethod def execute(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]: """Führt eine Aktion mit den gegebenen Parametern aus. Args: action: Die auszuführende Aktion (z.B. 'create_campaign'). params: Ein Dictionary mit Parametern für die Aktion. Returns: Ein Dictionary mit dem Ergebnis der Aktion. """ pass def validate_params(self, action: str, params: Dict[str, Any]) -> bool: """Validiert die Parameter für eine Aktion. Kann in Subklassen überschrieben werden. """ return True ``` **📌 Schritt 2: Google Ads Adapter implementieren (3 Tage)** 1. **Google Ads API-Zugriff einrichten**: - Folge der Anleitung: [Google Ads API Start](https://developers.google.com/google-ads/api/docs/start). - Erstelle ein **Entwicklerkonto** und lade die `google-ads.yaml`-Datei herunter. - Installiere die Client-Bibliothek: ```bash pip install google-ads-api ``` 2. Erstelle `src/adapters/google_ads_adapter.py`: ```python from google.ads.googleads.client import GoogleAdsClient from google.ads.googleads.errors import GoogleAdsException from typing import Dict, Any, List from .base_adapter import ToolAdapter class GoogleAdsAdapter(ToolAdapter): def __init__(self, yaml_path: str = "google-ads.yaml"): self.client = GoogleAdsClient.load_from_storage(yaml_path) self.customer_id = None # Wird später gesetzt def get_name(self) -> str: return "google_ads" def get_actions(self) -> List[str]: return [ "create_campaign", "get_campaigns", "create_ad_group", "get_ad_groups", "create_ad", "get_ads" ] def set_customer_id(self, customer_id: str): """Setzt die Kunden-ID für alle zukünftigen Anfragen.""" self.customer_id = customer_id def execute(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]: if not self.customer_id: raise ValueError("Customer ID nicht gesetzt. Nutze set_customer_id().") if action == "create_campaign": return self._create_campaign(params) elif action == "get_campaigns": return self._get_campaigns(params) elif action == "create_ad_group": return self._create_ad_group(params) elif action == "get_ad_groups": return self._get_ad_groups(params) elif action == "create_ad": return self._create_ad(params) elif action == "get_ads": return self._get_ads(params) else: raise ValueError(f"Unbekannte Aktion: {action}") def _create_campaign(self, params: Dict[str, Any]) -> Dict[str, Any]: """Erstellt eine neue Kampagne in Google Ads.""" campaign_service = self.client.get_service("CampaignService") campaign = self.client.get_type("Campaign") campaign.name = params.get("name", "Neue Kampagne") campaign.campaign_budget = self._create_budget(params.get("budget", 1000000)) # 1.000.000 Mikro-Euro = 1€ campaign.advertising_channel_type = self.client.get_type("AdvertisingChannelTypeEnum").AdvertisingChannelType.MULTI_CHANNEL campaign.status = self.client.get_type("CampaignStatusEnum").CampaignStatus.PAUSED # Zielgruppe (optional) if "targeting" in params: campaign.targeting_setting = self._create_targeting_setting(params["targeting"]) # Zeitplan (optional) if "start_date" in params or "end_date" in params: campaign.date_range = self._create_date_range( params.get("start_date"), params.get("end_date") ) try: campaign_response = campaign_service.create_campaign( customer_id=self.customer_id, campaign=campaign ) return { "status": "success", "campaign_id": campaign_response.resource_name, "name": campaign.name } except GoogleAdsException as ex: return { "status": "error", "error": str(ex.error.code()), "message": ex.error.message() } def _create_budget(self, amount: float) -> Any: """Erstellt ein Kampagnen-Budget.""" budget_service = self.client.get_service("CampaignBudgetService") budget = self.client.get_type("CampaignBudget") budget.name = f"Budget für {amount}€" budget.amount_micros = int(amount * 1_000_000) # Umrechnung in Mikro-Euro budget.delivery_method = self.client.get_type("BudgetDeliveryMethodEnum").BudgetDeliveryMethod.STANDARD budget.status = self.client.get_type("BudgetStatusEnum").BudgetStatus.ENABLED return budget_service.create_campaign_budget( customer_id=self.customer_id, campaign_budget=budget ).resource_name def _create_targeting_setting(self, targeting: Dict[str, Any]) -> Any: """Erstellt Einstellungen für die Zielgruppe.""" targeting_setting = self.client.get_type("TargetingSetting") if "locations" in targeting: targeting_setting.geo_target_type = self.client.get_type("GeoTargetTypeEnum").GeoTargetType.PRESENCE targeting_setting.geo_targets = [ self.client.get_service("GeographicViewService").geographic_view_path( location_id=loc_id ) for loc_id in targeting["locations"] ] return targeting_setting def _create_date_range(self, start_date: str = None, end_date: str = None) -> Any: """Erstellt einen Zeitrahmen für die Kampagne.""" date_range = self.client.get_type("DateRange") if start_date: date_range.start_date = start_date if end_date: date_range.end_date = end_date return date_range def _get_campaigns(self, params: Dict[str, Any]) -> Dict[str, Any]: """Holt alle Kampagnen für den aktuellen Kunden.""" campaign_service = self.client.get_service("CampaignService") query = f""" SELECT campaign.id, campaign.name, campaign.status, campaign.start_date, campaign.end_date FROM campaign WHERE campaign.advertising_channel_type = 'MULTI_CHANNEL' """ results = campaign_service.list_campaigns( customer_id=self.customer_id, query=query ) campaigns = [] for row in results: campaigns.append({ "id": row.campaign.id, "name": row.campaign.name, "status": row.campaign.status, "start_date": row.campaign.start_date, "end_date": row.campaign.end_date }) return {"status": "success", "campaigns": campaigns} def _create_ad_group(self, params: Dict[str, Any]) -> Dict[str, Any]: """Erstellt eine Anzeigengruppe.""" ad_group_service = self.client.get_service("AdGroupService") ad_group = self.client.get_type("AdGroup") ad_group.name = params.get("name", "Neue Anzeigengruppe") ad_group.status = self.client.get_type("AdGroupStatusEnum").AdGroupStatus.ENABLED ad_group.campaign = campaign_service.campaign_path( self.customer_id, params["campaign_id"] ) ad_group.type = self.client.get_type("AdGroupTypeEnum").AdGroupType.SEARCH_STANDARD # Zielgruppe (optional) if "targeting" in params: ad_group.targeting_setting = self._create_targeting_setting(params["targeting"]) try: ad_group_response = ad_group_service.create_ad_group( customer_id=self.customer_id, ad_group=ad_group ) return { "status": "success", "ad_group_id": ad_group_response.resource_name, "name": ad_group.name } except GoogleAdsException as ex: return { "status": "error", "error": str(ex.error.code()), "message": ex.error.message() } def _get_ad_groups(self, params: Dict[str, Any]) -> Dict[str, Any]: """Holt alle Anzeigengruppen für eine Kampagne.""" ad_group_service = self.client.get_service("AdGroupService") campaign_id = params.get("campaign_id") if not campaign_id: return {"status": "error", "message": "campaign_id ist erforderlich."} query = f""" SELECT ad_group.id, ad_group.name, ad_group.status FROM ad_group WHERE ad_group.campaign = '{campaign_service.campaign_path(self.customer_id, campaign_id)}' """ results = ad_group_service.list_ad_groups( customer_id=self.customer_id, query=query ) ad_groups = [] for row in results: ad_groups.append({ "id": row.ad_group.id, "name": row.ad_group.name, "status": row.ad_group.status }) return {"status": "success", "ad_groups": ad_groups} def _create_ad(self, params: Dict[str, Any]) -> Dict[str, Any]: """Erstellt eine Anzeige.""" ad_group_ad_service = self.client.get_service("AdGroupAdService") ad = self.client.get_type("Ad") ad.final_urls = [params.get("final_url", "https://example.com")] ad.status = self.client.get_type("AdStatusEnum").AdStatus.ENABLED # Anzeigentyp (z.B. Textanzeige) if params.get("type") == "text": text_ad = self.client.get_type("TextAdInfo") text_ad.headline1 = params.get("headline1", "Kopfzeile 1") text_ad.headline2 = params.get("headline2", "Kopfzeile 2") text_ad.headline3 = params.get("headline3", "Kopfzeile 3") text_ad.description1 = params.get("description1", "Beschreibung 1") text_ad.description2 = params.get("description2", "Beschreibung 2") text_ad.path1 = params.get("path1", "Pfad 1") text_ad.path2 = params.get("path2", "Pfad 2") ad.text_ad = text_ad try: ad_response = ad_group_ad_service.create_ad_group_ad( customer_id=self.customer_id, ad_group_id=params["ad_group_id"], ad=ad ) return { "status": "success", "ad_id": ad_response.resource_name, "type": "text" } except GoogleAdsException as ex: return { "status": "error", "error": str(ex.error.code()), "message": ex.error.message() } def _get_ads(self, params: Dict[str, Any]) -> Dict[str, Any]: """Holt alle Anzeigen für eine Anzeigengruppe.""" ad_group_ad_service = self.client.get_service("AdGroupAdService") ad_group_id = params.get("ad_group_id") if not ad_group_id: return {"status": "error", "message": "ad_group_id ist erforderlich."} query = f""" SELECT ad_group_ad.ad.id, ad_group_ad.ad.type, ad_group_ad.ad.final_urls, ad_group_ad.status FROM ad_group_ad WHERE ad_group_ad.ad_group = '{ad_group_ad_service.ad_group_path(self.customer_id, ad_group_id)}' """ results = ad_group_ad_service.list_ad_group_ads( customer_id=self.customer_id, query=query ) ads = [] for row in results: ads.append({ "id": row.ad_group_ad.ad.id, "type": row.ad_group_ad.ad.type, "final_urls": row.ad_group_ad.ad.final_urls, "status": row.ad_group_ad.status }) return {"status": "success", "ads": ads} ``` **📌 Schritt 3: Tool-Agent erstellen (2 Tage)** Erstelle `src/agents/tool_agent.py`: ```python from typing import Dict, Any, Optional from src.adapters.base_adapter import ToolAdapter from src.adapters.google_ads_adapter import GoogleAdsAdapter class ToolAgent: def __init__(self): self.adapters: Dict[str, ToolAdapter] = { "google_ads": GoogleAdsAdapter() } def get_available_tools(self) -> Dict[str, list]: """Gibt eine Liste aller verfügbaren Tools und Aktionen zurück.""" return { name: adapter.get_actions() for name, adapter in self.adapters.items() } def execute( self, tool_name: str, action: str, params: Dict[str, Any], customer_id: Optional[str] = None ) -> Dict[str, Any]: """Führt eine Aktion mit einem Tool aus.""" adapter = self.adapters.get(tool_name) if not adapter: return {"status": "error", "message": f"Tool {tool_name} nicht gefunden."} if customer_id and hasattr(adapter, "set_customer_id"): adapter.set_customer_id(customer_id) try: return adapter.execute(action, params) except Exception as e: return {"status": "error", "message": str(e)} def add_adapter(self, name: str, adapter: ToolAdapter): """Fügt einen neuen Adapter hinzu.""" self.adapters[name] = adapter ``` **📌 Schritt 4: Orchestrator-Agent erweitern (3 Tage)** Aktualisiere `src/agents/orchestrator.py`: ```python from typing import Dict, Any, Optional, List from src.rag import query_rag from src.neo4j_memory import Neo4jMemory from src.context import ContextManager from src.agents.tool_agent import ToolAgent import re import json class OrchestratorAgent: def __init__(self): self.rag = query_rag self.memory = Neo4jMemory( uri="bolt://localhost:7687", user="neo4j", password="password" ) self.memory._ensure_indexes() self.context = ContextManager() self.tool_agent = ToolAgent() def handle_query( self, user_id: str, session_id: str, query: str, customer_id: Optional[str] = None ) -> str: # 1. Kontext laden context = self.context.get_context(session_id) if not context: context = {"user_id": user_id, "session_id": session_id} if customer_id: context["current_customer"] = customer_id # 2. Erinnerung laden memories = self.memory.get_last_interactions(user_id, limit=3) memory_context = "\n".join( [f"[{m['time']}] {m['agent_id']}: {m['message']}" for m in memories] ) # 3. Kundeninfo laden customer_info = {} if customer_id: customer_info = self.memory.get_customer_info(customer_id) context["current_customer"] = customer_id # 4. Tool-Nutzung prüfen tool_result = None if self._should_use_tool(query): tool_result = self._use_tool(query, customer_id) if tool_result and tool_result.get("status") == "success": # Speichere Tool-Ergebnis in Erinnerung self.memory.add_interaction( user_id=user_id, agent_id="tool_agent", message=f"Tool-Aktion: {tool_result}", customer_id=customer_id ) # 5. Kontext für RAG vorbereiten augmented_query = f""" Frage: {query} Kontext: - Nutzer: {user_id} - Session: {session_id} - Aktueller Kunde: {context.get('current_customer', 'keiner')} - Letzte Interaktionen: {memory_context} - Kundeninfo: {customer_info} - Tool-Ergebnis: {tool_result} """ # 6. RAG abfragen (falls kein Tool direkt geantwortet hat) if not tool_result or tool_result.get("status") != "success": response = self.rag(augmented_query) else: response = f"Tool-Aktion erfolgreich: {tool_result.get('message', tool_result)}" # 7. Erinnerung speichern self.memory.add_interaction( user_id=user_id, agent_id="orchestrator", message=query, customer_id=customer_id ) # 8. Kontext aktualisieren self.context.update_context(session_id, { "last_query": query, "last_response": response, "last_tool_result": tool_result }) return response def _should_use_tool(self, query: str) -> bool: """Prüft, ob die Query eine Tool-Nutzung erfordert.""" tool_keywords = [ "erstelle kampagne", "create campaign", "anzeige erstellen", "create ad", "anzeigengruppe erstellen", "create ad group", "hole kampagnen", "get campaigns" ] return any(keyword in query.lower() for keyword in tool_keywords) def _use_tool(self, query: str, customer_id: Optional[str] = None) -> Optional[Dict[str, Any]]: """Führt eine Tool-Aktion basierend auf der Query aus.""" query_lower = query.lower() # Beispiel: "Erstelle eine Kampagne für Kunde Acme mit Budget 5000€" if "erstelle kampagne" in query_lower or "create campaign" in query_lower: params = self._extract_campaign_params(query) return self.tool_agent.execute( tool_name="google_ads", action="create_campaign", params=params, customer_id=customer_id ) elif "hole kampagnen" in query_lower or "get campaigns" in query_lower: return self.tool_agent.execute( tool_name="google_ads", action="get_campaigns", params={}, customer_id=customer_id ) elif "erstelle anzeige" in query_lower or "create ad" in query_lower: params = self._extract_ad_params(query) return self.tool_agent.execute( tool_name="google_ads", action="create_ad", params=params, customer_id=customer_id ) else: return None def _extract_campaign_params(self, query: str) -> Dict[str, Any]: """Extrahiert Parameter für eine Kampagne aus der Query.""" params = {} query_lower = query.lower() # Budget extrahieren (z.B. "Budget 5000€") budget_match = re.search(r"budget[:\s]+(\d+)\s*(€|eur|euro)?", query_lower, re.IGNORECASE) if budget_match: params["budget"] = float(budget_match.group(1)) # Name extrahieren (z.B. "Name: Sommer-Kampagne 2024") name_match = re.search(r"name[:\s]+([^\n,;]+)", query_lower, re.IGNORECASE) if name_match: params["name"] = name_match.group(1).strip() # Startdatum extrahieren (z.B. "Start: 2024-06-01") start_match = re.search(r"start[:\s]+(\d{4}-\d{2}-\d{2})", query_lower) if start_match: params["start_date"] = start_match.group(1) # Enddatum extrahieren end_match = re.search(r"ende[:\s]+(\d{4}-\d{2}-\d{2})", query_lower) if end_match: params["end_date"] = end_match.group(1) # Standardwerte if "name" not in params: params["name"] = "Neue Kampagne" if "budget" not in params: params["budget"] = 1000 # Standard-Budget: 1000€ return params def _extract_ad_params(self, query: str) -> Dict[str, Any]: """Extrahiert Parameter für eine Anzeige aus der Query.""" params = {} query_lower = query.lower() # Anzeigengruppen-ID (erforderlich) ad_group_match = re.search(r"anzeigengruppe[:\s]+([a-zA-Z0-9_-]+)", query_lower) if ad_group_match: params["ad_group_id"] = ad_group_match.group(1) # Final URL url_match = re.search(r"url[:\s]+(https?://[^\s]+)", query_lower) if url_match: params["final_url"] = url_match.group(1) # Headlines headline_matches = re.findall(r"kopfzeile\s*(\d+)[:\s]+([^\n,;]+)", query_lower, re.IGNORECASE) for i, match in enumerate(headline_matches[:3], 1): # Max. 3 Headlines params[f"headline{i}"] = match[1].strip() # Descriptions desc_matches = re.findall(r"beschreibung\s*(\d+)[:\s]+([^\n,;]+)", query_lower, re.IGNORECASE) for i, match in enumerate(desc_matches[:2], 1): # Max. 2 Descriptions params[f"description{i}"] = match[1].strip() # Typ (Standard: text) params["type"] = "text" return params ``` **📌 Schritt 5: Frontend für Tool-Nutzung erweitern (1 Tag)** Aktualisiere `app.py`: ```python import streamlit as st from src.agents.orchestrator import OrchestratorAgent from src.neo4j_memory import Neo4jMemory from src.agents.tool_agent import ToolAgent # Initialisiere Agenten orchestrator = OrchestratorAgent() memory = Neo4jMemory("bolt://localhost:7687", "neo4j", "password") tool_agent = ToolAgent() # UI st.title("🚀 Werbeagentur Agent (Phase 3)") user_id = st.text_input("Nutzer-ID", value="user_1") session_id = st.text_input("Session-ID", value="session_1") customer_id = st.text_input("Kunden-ID (optional)", value="acme") # Tool-Auswahl st.subheader("🔧 Tool-Nutzung") use_tool = st.checkbox("Tool direkt nutzen (überspringt RAG)") if use_tool: col1, col2 = st.columns(2) with col1: tool_name = st.selectbox( "Tool auswählen", options=list(tool_agent.get_available_tools().keys()) ) with col2: if tool_name: actions = tool_agent.get_available_tools()[tool_name] action = st.selectbox("Aktion auswählen", options=actions) tool_params = st.text_area( "Parameter (JSON)", value='{"name": "Test-Kampagne", "budget": 5000}' ) query = st.text_input("Frage stellen:") if st.button("Senden"): with st.spinner("Denke nach..."): if use_tool and tool_name and action: try: params = json.loads(tool_params) if tool_params else {} result = tool_agent.execute( tool_name=tool_name, action=action, params=params, customer_id=customer_id if customer_id else None ) st.write("🤖 **Tool-Ergebnis:**", result) except Exception as e: st.error(f"Fehler bei Tool-Nutzung: {e}") else: response = orchestrator.handle_query( user_id=user_id, session_id=session_id, query=query, customer_id=customer_id if customer_id else None ) st.write("🤖 **Antwort:**", response) # Erinnerung anzeigen st.subheader("📜 Letzte Interaktionen:") memories = memory.get_last_interactions(user_id, limit=5) for mem in memories: st.write(f"**{mem['agent_id']}** ({mem['time']}): {mem['message']}") # Kundeninfo anzeigen if customer_id: st.subheader("👤 Kundeninfo:") customer_info = memory.get_customer_info(customer_id) st.json(customer_info) ``` → **Ergebnis**: Agenten können jetzt **Google Ads Kampagnen erstellen**! --- --- ### **📌 Phase 4: Skalierung & Optimierung (2 Wochen)** **Ziel**: Performance verbessern, mehr Agenten hinzufügen, Monitoring einrichten. #### **🔧 Tech-Stack** | Komponente | Technologie | Kosten | |---------------------|---------------------------|---------| | **Caching** | Redis | 0€ | | **Asynchrone Aufgaben** | Celery + RabbitMQ | 0€ | | **Monitoring** | Prometheus + Grafana | 0€ | | **Mehr Agenten** | CrewAI | 0€ | --- #### **🛠 Schritt-für-Schritt-Anleitung** **📌 Schritt 1: Caching für RAG einrichten (1 Tag)** Aktualisiere `src/rag.py`: ```python from langchain.cache import RedisCache import redis # Redis für Caching einrichten redis_client = redis.Redis(host='localhost', port=6379, db=0) cache = RedisCache(redis_=redis_client) # RAG-Kette mit Caching qa = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=db.as_retriever(search_kwargs={"k": 3}), cache=cache, # Caching aktivieren return_source_documents=True ) ``` **📌 Schritt 2: Asynchrone Verarbeitung mit Celery (2 Tage)** 1. Installiere Celery und RabbitMQ: ```bash pip install celery redis docker run -d --name rabbitmq -p 5672:5672 rabbitmq:3 ``` 2. Erstelle `tasks.py`: ```python from celery import Celery from src.agents.orchestrator import OrchestratorAgent app = Celery( 'ad_agent_tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0' ) orchestrator = OrchestratorAgent() @app.task(bind=True) def handle_query_async(self, user_id: str, session_id: str, query: str, customer_id: str = None): return orchestrator.handle_query( user_id=user_id, session_id=session_id, query=query, customer_id=customer_id ) ``` 3. Starte Celery: ```bash celery -A tasks worker --loglevel=info ``` 4. Aktualisiere `app.py` für asynchrone Verarbeitung: ```python from tasks import handle_query_async from celery.result import AsyncResult # ... if st.button("Senden"): with st.spinner("Wird verarbeitet..."): if use_tool and tool_name and action: # ... (Tool-Nutzung wie vorher) else: # Asynchrone Verarbeitung task = handle_query_async.delay( user_id=user_id, session_id=session_id, query=query, customer_id=customer_id if customer_id else None ) st.write("⏳ Aufgabe gestartet. ID:", task.id) # Fortschritt anzeigen progress_bar = st.progress(0) status_text = st.empty() for i in range(100): result = AsyncResult(task.id) if result.ready(): response = result.get() st.write("🤖 **Antwort:**", response) break else: progress_bar.progress((i % 20) / 20) status_text.text(f"Warte auf Ergebnis... ({i}%)") time.sleep(0.1) ``` **📌 Schritt 3: Mehr Agenten hinzufügen (3 Tage)** Installiere CrewAI: ```bash pip install crewai ``` Erstelle `src/agents/creative_agent.py`: ```python from crewai import Agent from langchain.llms import Ollama class CreativeAgent: def __init__(self): self.llm = Ollama(model="mistral") self.agent = Agent( role="Kreativer Werbetexter", goal="Erstelle überzeugende und kreative Werbetexte für Kampagnen.", backstory=""" Du bist ein erfahrener Werbetexter mit 10 Jahren Erfahrung in der Branche. Du hast für große Marken wie Nike, Coca-Cola und Apple gearbeitet. Deine Texte sind immer kreativ, ansprechend und zielgruppengerecht. """, verbose=True, allow_delegation=False, llm=self.llm ) def generate_ideas(self, prompt: str, context: Dict[str, Any]) -> str: """Generiert kreative Ideen für eine Kampagne.""" task = self.agent.create_task( description=f""" Kontext: {context} Aufgabe: {prompt} Generiere 3 kreative Ideen für eine Werbekampagne. Jede Idee sollte enthalten: - einen Namen - eine kurze Beschreibung - die Zielgruppe - den einzigartigen Verkaufspunkt (USP) """, expected_output="Eine Liste von 3 kreativen Kampagnenideen im JSON-Format." ) result = self.agent.execute_task(task) return result def generate_copy(self, prompt: str, context: Dict[str, Any]) -> str: """Generiert Werbetexte (z.B. für Anzeigen).""" task = self.agent.create_task( description=f""" Kontext: {context} Aufgabe: {prompt} Schreibe einen überzeugenden Werbetext für eine Anzeige. Der Text sollte enthalten: - Eine fesselnde Headline - 2-3 Sätze Beschreibung - Ein klarer Call-to-Action (CTA) """, expected_output="Ein Werbetext mit Headline, Beschreibung und CTA." ) result = self.agent.execute_task(task) return result ``` Aktualisiere `src/agents/orchestrator.py`: ```python from src.agents.creative_agent import CreativeAgent class OrchestratorAgent: def __init__(self): # ... (vorheriger Code) self.creative_agent = CreativeAgent() def handle_query(self, ...): # ... (vorheriger Code) # 5. Kreativagent nutzen (falls nötig) if "ideen" in query_lower or "kreativ" in query_lower: ideas = self.creative_agent.generate_ideas(query, context) response += f"\n\n💡 **Kreative Ideen:**\n{ideas}" elif "text" in query_lower or "anzeige" in query_lower: copy = self.creative_agent.generate_copy(query, context) response += f"\n\n✍️ **Werbetext:**\n{copy}" # ... (restlicher Code) ``` **📌 Schritt 4: Monitoring mit Prometheus & Grafana (2 Tage)** 1. Installiere Prometheus und Grafana: ```bash docker run -d -p 9090:9090 -v /path/to/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus docker run -d -p 3000:3000 grafana/grafana ``` 2. Erstelle `prometheus.yml`: ```yaml global: scrape_interval: 15s scrape_configs: - job_name: 'ad_agent' static_configs: - targets: ['localhost:8000'] # FastAPI Port ``` 3. Füge Metriken zu FastAPI hinzu (`src/monitoring.py`): ```python from fastapi import FastAPI from prometheus_fastapi_instrumentator import Instrumentator from prometheus_client import Counter, Histogram app = FastAPI() # Metriken definieren QUERY_COUNTER = Counter( 'agent_queries_total', 'Anzahl der Anfragen an den Agenten', ['user_id', 'agent_type'] ) QUERY_DURATION = Histogram( 'agent_query_duration_seconds', 'Dauer der Agenten-Anfragen in Sekunden', ['user_id', 'agent_type'] ) # Instrumentator für FastAPI Instrumentator().instrument(app).expose(app) # Beispiel: Metriken in einer Route aktualisieren @app.get("/metrics-example") def metrics_example(): with QUERY_DURATION.labels(user_id="test", agent_type="orchestrator").time(): QUERY_COUNTER.labels(user_id="test", agent_type="orchestrator").inc() return {"message": "Metrics updated"} ``` 4. Integriere Metriken in den Orchestrator: ```python from src.monitoring import QUERY_COUNTER, QUERY_DURATION class OrchestratorAgent: def handle_query(self, user_id, session_id, query, customer_id=None): with QUERY_DURATION.labels(user_id=user_id, agent_type="orchestrator").time(): QUERY_COUNTER.labels(user_id=user_id, agent_type="orchestrator").inc() # ... (restlicher Code) ``` **📌 Schritt 5: Deployment vorbereiten (2 Tage)** 1. Erstelle `Dockerfile`: ```dockerfile FROM python:3.10-slim WORKDIR /app # Abhängigkeiten installieren COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # Code kopieren COPY . . # Ollama installieren (für lokale LLMs) RUN curl -fsSL https://ollama.ai/install.sh | sh # Port exponieren EXPOSE 8000 # Startbefehl CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"] ``` 2. Erstelle `docker-compose.yml`: ```yaml version: '3.8' services: backend: build: . ports: - "8000:8000" environment: - NEO4J_URI=bolt://neo4j:7687 - NEO4J_USER=neo4j - NEO4J_PASSWORD=password - REDIS_HOST=redis depends_on: - neo4j - redis - rabbitmq volumes: - ./data:/app/data # Dokumente für RAG - ./faiss_index:/app/faiss_index # Vektordatenbank neo4j: image: neo4j:community ports: - "7474:7474" - "7687:7687" environment: NEO4J_AUTH: neo4j/password NEO4J_dbms_memory_heap_max__size: 2G volumes: - neo4j_data:/data - neo4j_logs:/logs - neo4j_import:/var/lib/neo4j/import - neo4j_plugins:/plugins redis: image: redis:alpine ports: - "6379:6379" volumes: - redis_data:/data rabbitmq: image: rabbitmq:3 ports: - "5672:5672" - "15672:15672" # Management-UI volumes: - rabbitmq_data:/var/lib/rabbitmq prometheus: image: prom/prometheus ports: - "9090:9090" volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml grafana: image: grafana/grafana ports: - "3000:3000" volumes: - grafana_data:/var/lib/grafana depends_on: - prometheus volumes: neo4j_data: neo4j_logs: neo4j_import: neo4j_plugins: redis_data: rabbitmq_data: grafana_data: ``` 3. Starte alles mit: ```bash docker-compose up -d --build ``` → **Ergebnis**: Ein **skalierbares, überwachtes System** mit mehreren Agenten! --- --- ### **📌 Phase 5: Deployment (1 Woche)** **Ziel**: System in der Cloud oder lokal produktiv setzen. #### **🔧 Option 1: Lokale Bereitstellung (für Tests/Entwicklung)** - Nutze `docker-compose up -d` (wie oben). - **Kosten**: 0€ (außer Hardware). #### **🔧 Option 2: Cloud-Bereitstellung (für Produktion)** 1. **Server auswählen**: - **Hetzner Cloud**: ~5–20€/Monat (je nach Größe). - Beispiel: CX21 (2 vCPUs, 4GB RAM) für ~7€/Monat. - **Oracle Cloud Free Tier**: Kostenlos (aber komplexer). 2. **Deployment-Schritte**: ```bash # 1. Server erstellen (Ubuntu 22.04) # 2. Docker und Docker Compose installieren sudo apt update && sudo apt install -y docker.io docker-compose # 3. Code auf den Server kopieren (z.B. mit Git oder scp) git clone && cd # 4. Docker-Images bauen und starten docker-compose up -d --build # 5. Domain und HTTPS einrichten (optional) sudo apt install -y nginx certbot python3-certbot-nginx sudo certbot --nginx -d deine-domain.de ``` 3. **Monitoring einrichten**: - Greife auf Grafana zu: `http://:3000` (Standard-Login: `admin/admin`). - Konfiguriere Dashboards für: - Anfragen pro Minute - Antwortzeiten - Fehlerraten --- --- --- ## 💰 **Kostenübersicht (Monatlich)** | Komponente | Kosten (€) | Anmerkungen | |---------------------|-------------|----------------------------------| | **Server (Hetzner CX21)** | 7,49 | 2 vCPUs, 4GB RAM, 40GB SSD | | **LLMs (lokal)** | 0 | Ollama + Mistral-7B | | **Neo4j** | 0 | Community Edition (self-hosted) | | **Redis** | 0 | Self-hosted | | **RabbitMQ** | 0 | Self-hosted | | **Prometheus/Grafana** | 0 | Self-hosted | | **Google Ads API** | 0 | Kostenlos für Entwickler | | **Domain + HTTPS** | ~1–5 | Optional (z.B. bei Namecheap) | | **Gesamt** | **8–12€** | | *Hinweis*: Falls du Cloud-LLMs (z.B. OpenAI) nutzt, kommen **~0,01–0,10€ pro Anfrage** hinzu. --- --- --- ## ⏳ **Zeitplan (11 Wochen)** | Phase | Dauer | Aufgaben | Ergebnis | |---------------------|----------|--------------------------------------------------------------------------|-----------------------------------| | **Phase 1: MVP** | 2 Wochen | RAG + einfache Erinnerung (JSON) + Streamlit-Frontend | Funktionierendes Grundsystem | | **Phase 2: Erinnerung & Kontext** | 3 Wochen | Neo4j + Redis + Orchestrator-Agent | Dynamische Erinnerung & Kontext | | **Phase 3: Tool-Integration** | 3 Wochen | Tool-Adapter (Google Ads) + Tool-Agent + Frontend-Erweiterung | Agenten nutzen externe Tools | | **Phase 4: Skalierung** | 2 Wochen | Caching, Celery, mehr Agenten (CrewAI), Monitoring | Skalierbares, überwachtes System | | **Phase 5: Deployment** | 1 Woche | Docker, Cloud-Setup, HTTPS | Produktivsystem | **Gesamt**: **11 Wochen** (mit 1 Entwickler Vollzeit). Mit **2 Entwicklern** könnte es in **~6 Wochen** geschafft werden. --- --- --- ## 📁 **Projektstruktur (Empfohlen)** ``` ad_agent/ ├── data/ # Dokumente für RAG │ ├── kampagnen.pdf │ ├── kundeninfos.docx │ └── ... ├── faiss_index/ # Vektordatenbank (wird automatisch erstellt) ├── src/ │ ├── agents/ # Agenten-Klassen │ │ ├── orchestrator.py │ │ ├── creative_agent.py │ │ ├── knowledge_agent.py │ │ └── tool_agent.py │ ├── adapters/ # Tool-Adapter │ │ ├── base_adapter.py │ │ ├── google_ads_adapter.py │ │ └── canva_adapter.py │ ├── utils/ # Hilfsfunktionen │ │ ├── rag.py │ │ ├── neo4j_memory.py │ │ ├── context.py │ │ └── monitoring.py │ └── main.py # FastAPI/Streamlit App ├── tasks.py # Celery-Aufgaben ├── app.py # Streamlit-Frontend ├── docker-compose.yml # Docker-Konfiguration ├── Dockerfile # Docker-Image ├── requirements.txt # Python-Abhängigkeiten ├── prometheus.yml # Prometheus-Konfiguration └── README.md # Projektbeschreibung ``` --- --- --- ## 🎯 **Erste Schritte (Diese Woche)** 1. **✅ Dokumente sammeln** (10–20 relevante Dateien für RAG). 2. **✅ Lokale Umgebung einrichten**: - Python 3.10+ installieren. - Docker installieren. - Ollama + Mistral-7B installieren (`ollama pull mistral`). 3. **✅ Phase 1 umsetzen**: - `src/rag.py` erstellen (RAG mit FAISS). - `src/memory.py` erstellen (einfache JSON-Erinnerung). - `app.py` erstellen (Streamlit-Frontend). - System testen: `streamlit run app.py`. --- --- --- ## 🚀 **Langfristige Optimierungen (Nach dem MVP)** | **Optimierung** | **Zweck** | **Technologie** | |----------------------------|---------------------------------------------------------------------------|-------------------------------| | **Hybrid-Suche** | Bessere RAG-Ergebnisse durch Kombination von Vektor- und Keyword-Suche. | Elasticsearch, Weaviate | | **Event Sourcing** | Vollständige Historie aller Interaktionen. | Kafka, EventStoreDB | | **Multi-Modalität** | Bilder/Videos verarbeiten (z.B. für Social Media). | CLIP, Stable Diffusion | | **Echtzeit-Daten** | Live-Daten von Google Analytics, etc. | Webhooks, Kafka | | **Benutzerfeedback** | System durch Feedback verbessern. | RLHF, LangChain Feedback | | **Sicherheit** | Authentifizierung, Berechtigungen. | FastAPI Users, OAuth2 | | **Skalierbarkeit** | Für viele Nutzer optimieren. | Kubernetes, Load Balancing | --- --- --- ## 🔗 **Empfohlene Ressourcen** ### **Tutorials & Dokus** | **Thema** | **Ressource** | **Link** | |-------------------------|------------------------------------------------------------------------------|-----------------------------------------------| | **RAG** | LangChain RAG Tutorial | [Link](https://python.langchain.com/docs/use_cases/question_answering/) | | **Neo4j** | Neo4j Python Driver Docs | [Link](https://neo4j.com/docs/python-manual/current/) | | **Agenten** | CrewAI Docs | [Link](https://docs.crewai.com/) | | **Google Ads API** | Offizielle API-Dokumentation | [Link](https://developers.google.com/google-ads/api/docs/start) | | **Deployment** | Docker Compose Guide | [Link](https://docs.docker.com/compose/) | ### **Beispielprojekte** - [AutoGen Multi-Agenten-Chat](https://github.com/microsoft/autogen/tree/main/samples) - [LangChain + Neo4j](https://github.com/langchain-ai/langchain/tree/master/libs/langchain/langchain/graphs) - [CrewAI Beispiel](https://github.com/joaomdmoura/crewAI/tree/main/examples) ### **Community** - [LangChain Discord](https://discord.gg/80R5K7HxWV) - [Neo4j Community](https://community.neo4j.com/) - [CrewAI Discord](https://discord.gg/crewai) --- --- --- ## ⚠️ **Häufige Fallstricke & Lösungen** | **Problem** | **Lösung** | |--------------------------------------|----------------------------------------------------------------------------| | RAG gibt irrelevante Ergebnisse | Nutze bessere Embeddings (z.B. `bge-base-en-v1.5`) oder Hybrid-Suche. | | Neo4j ist langsam | Optimierte Abfragen schreiben, Indizes erstellen. | | LLMs sind zu langsam | Nutze kleinere Modelle (z.B. `mistral-7b-instruct`) oder Quantisierung. | | Tool-Adapter sind komplex | Beginne mit einfachen Tools (z.B. Mock-API für Tests). | | Erinnerung ist unvollständig | Speichere alle Interaktionen als Events in Neo4j. | | Kontext geht verloren | Nutze Session-IDs und speichere Kontext in Redis. | | Skalierbarkeit | Nutze asynchrone Verarbeitung (Celery) und Caching (Redis). | --- --- --- ## 🎉 **Zusammenfassung: Dein Aktionsplan** ### **📌 Diese Woche (Phase 1: MVP)** | **Tag** | **Aufgabe** | **Ergebnis** | |---------|-----------------------------------------------------------------------------|---------------------------------------| | 1 | Dokumente sammeln + Umgebung einrichten (Python, Docker, Ollama). | Bereites Entwicklungssystem | | 2 | RAG-System mit FAISS + Sentence-Transformers bauen. | `src/rag.py` | | 3 | Einfache Erinnerung (JSON) implementieren. | `src/memory.py` | | 4 | Streamlit-Frontend erstellen. | `app.py` | | 5 | System testen und erste Fragen stellen. | Funktionierendes MVP! | ### **📌 Nächste Woche (Phase 2: Erinnerung & Kontext)** | **Tag** | **Aufgabe** | **Ergebnis** | |---------|-----------------------------------------------------------------------------|---------------------------------------| | 1 | Neo4j installieren und erste Daten laden. | Lokale Neo4j-Datenbank | | 2 | `Neo4jMemory`-Klasse implementieren. | `src/neo4j_memory.py` | | 3 | Redis für Kontext einrichten. | `src/context.py` | | 4 | Orchestrator-Agent erstellen. | `src/agents/orchestrator.py` | | 5 | Frontend aktualisieren. | Erweiterte `app.py` | ### **📌 Übernächste Woche (Phase 3: Tool-Integration)** | **Tag** | **Aufgabe** | **Ergebnis** | |---------|-----------------------------------------------------------------------------|---------------------------------------| | 1 | Basis-Adapter (`ToolAdapter`) erstellen. | `src/adapters/base_adapter.py` | | 2–4 | Google Ads Adapter implementieren. | `src/adapters/google_ads_adapter.py` | | 5 | Tool-Agent erstellen. | `src/agents/tool_agent.py` | | 6 | Orchestrator erweitern (Tool-Nutzung). | Aktualisierter `orchestrator.py` | | 7 | Frontend für Tool-Nutzung erweitern. | Aktualisierte `app.py` | --- --- --- ## 💡 **Finaler Rat** 1. **Starte klein und iteriere**: - Baue zuerst ein **MVP mit RAG + einfache Erinnerung** (Phase 1). - Füge dann schrittweise **Neo4j, Tools und mehr Agenten** hinzu. 2. **Nutze Open-Source-Tools**: - **CrewAI/AutoGen** für Agenten-Frameworks. - **FAISS/Neo4j** für Wissen und Erinnerung. - **Ollama** für lokale LLMs (kostengünstig). 3. **Dokumentiere alles**: - Besonders die **Tool-Adapter** und **Datenbank-Schemas**. - Nutze **Git** für Code und **DVC** für Daten. 4. **Teste mit echten Use Cases**: - Frage deine Kollegen, welche **konkreten Aufgaben** sie brauchen. - Baue diese **zuerst** (z.B. "Erstelle eine Google Ads Kampagne"). 5. **Kosten im Auge behalten**: - Nutze **lokale LLMs** (Ollama) statt Cloud-APIs. - Starte mit **kostenlosen Cloud-Tiers** (z.B. Hetzner für ~5€/Monat). --- --- --- --- ## 🚀 **Los geht’s!** Mit dieser Strategie kannst du **innerhalb von 11 Wochen** ein **voll funktionierendes Agentensystem** für deine Werbeagentur aufbauen – **kostengünstig (5–20€/Monat) und tool-unabhängig**. **Beginne diese Woche mit Phase 1 (MVP)** und arbeite dich schrittweise vor. Falls du Fragen hast oder Hilfe bei der Implementierung brauchst, lass es mich wissen! 🎯

Agentursystem