Natuurlijk! Hieronder vind je een uitgebreide gids en voorbeeldcode voor het ontwikkelen van een LangChain-agent die dynamisch kan interacteren met de OpenWeatherMap API en een aangepaste sentimentanalyse-tool. De code bevat methoden voor flexibele API-aanroepen, dynamische promptopbouw op basis van externe data, en het koppelen van meerdere tools voor verbeterde workflows. **Stap 1: Vereisten installeren** Zorg dat je de benodigde Python-pakketten hebt geïnstalleerd: ```bash pip install langchain openai requests ``` **Stap 2: Authenticatie en API-limieten** - **OpenWeatherMap API**: vereist API-sleutel, gratis plan tot 60 verzoeken/minuut. - **Sentimentanalyse Tool**: voorbeeld met een lokale of externe API met OAuth2. Max 1000 verzoeken/uur. **Stap 3: Implementatie** Hieronder een voorbeeld van een LangChain-agent die: - OAuth2-authenticatie gebruikt voor de sentimenttool. - Dynamisch API-aanroepen doet naar OpenWeatherMap. - Een aangepaste prompt opbouwt op basis van de API-respons. - Meerdere tools koppelt voor geavanceerde workflows. --- ### Voorbeeldcode ```python from langchain.agents import initialize_agent, Tool from langchain.llms import OpenAI import requests import time # API-configuraties OPENWEATHERMAP_API_KEY = "jouw_openweather_api_key" SENTIMENT_API_URL = "https://api.jouwsentimenttool.com/analyze" SENTIMENT_API_CLIENT_ID = "jouw_client_id" SENTIMENT_API_CLIENT_SECRET = "jouw_client_secret" # OAuth2-token ophalen voor sentimentanalyse tool def get_sentiment_token(): response = requests.post( "https://api.jouwsentimenttool.com/oauth/token", data={ "grant_type": "client_credentials", "client_id": SENTIMENT_API_CLIENT_ID, "client_secret": SENTIMENT_API_CLIENT_SECRET, } ) response.raise_for_status() return response.json()["access_token"] # Functie voor dynamische API-aanroepen naar OpenWeatherMap def get_weather(city): url = f"http://api.openweathermap.org/data/2.5/weather" params = { "q": city, "appid": OPENWEATHERMAP_API_KEY, "units": "metric" } response = requests.get(url, params=params) if response.status_code == 429: print("Rate limit bereikt bij OpenWeatherMap, wachten...") time.sleep(60) return get_weather(city) response.raise_for_status() data = response.json() weather_desc = data['weather'][0]['description'] temp = data['main']['temp'] return f"Het weer in {city} is {weather_desc} met een temperatuur van {temp}°C." # Tool voor weerinformatie def weather_tool(city): return get_weather(city) weather_tool_obj = Tool.from_function( func=weather_tool, name="WeatherTool", description="Geeft het weerbericht voor een opgegeven stad." ) # Tool voor sentimentanalyse def sentiment_analysis(text): token = get_sentiment_token() headers = { "Authorization": f"Bearer {token}" } payload = {"text": text} response = requests.post(SENTIMENT_API_URL, headers=headers, json=payload) if response.status_code == 429: print("Rate limit bereikt bij sentimentanalyse, wachten...") time.sleep(60) return sentiment_analysis(text) response.raise_for_status() result = response.json() sentiment = result.get("sentiment", "onbekend") confidence = result.get("confidence", 0) return f"Sentiment: {sentiment} (vertrouwen: {confidence*100:.1f}%)" sentiment_tool_obj = Tool.from_function( func=sentiment_analysis, name="SentimentAnalysisTool", description="Voert sentimentanalyse uit op tekst met OAuth2-authenticatie." ) # Initializeer de LLM (bijvoorbeeld OpenAI) llm = OpenAI(model="gpt-3.5-turbo", temperature=0) # Agent met tools tools = [weather_tool_obj, sentiment_tool_obj] # Voorbeeld van een aangepaste promptopbouw def custom_prompt(city): weather_info = weather_tool(city) prompt = ( f"Hier is de weersinformatie voor {city}:\n{weather_info}\n\n" "Schrijf een korte samenvatting en voer een sentimentanalyse uit op deze informatie." ) return prompt # Initialiseer de agent agent = initialize_agent(tools, llm, agent_type="zero-shot-react-description", verbose=True) # Voorbeeldgebruik city = "Amsterdam" prompt = custom_prompt(city) response = agent.run(prompt) print(response) ``` --- ### Beperkingen en overwegingen - **Rate limits**: - OpenWeatherMap: max 60 verzoeken per minuut (bij gratis plan). Implementatie met wachttijd bij 429. - Sentimenttool: max 1000 verzoeken per uur. Implementatie met wachttijd bij 429. - **Authenticatie**: - OpenWeatherMap: API-sleutel. - Sentimentanalyse: OAuth2 token opvragen via client credentials flow. - **Responsafhandeling**: - Foutafhandeling is opgenomen met `raise_for_status()` en retries bij rate limits. - **Flexibiliteit**: - Je kunt de `custom_prompt` aanpassen om dynamisch prompts te bouwen op basis van API-data. - Tools kunnen worden uitgebreid met extra functionaliteit. --- ### Samenvatting Deze aanpak maakt het mogelijk om een krachtige, dynamische LangChain-agent te bouwen die meerdere API's en tools integreert, met aandacht voor authenticatie, rate limits en flexibele promptopbouw. Pas de API-keys en URL's aan naar jouw configuratie en uitbreidingen voor jouw specifieke use-cases!

Hier is een uitgebreide implementatie van een LangChain-agent met de gevraagde integraties: ## 1. Installatievereisten ```bash pip install langchain openai python-dotenv requests ``` ## 2. Configuratiebestand (.env) ```env OPENAI_API_KEY=uw_openai_api_sleutel OPENWEATHER_API_KEY=uw_openweather_sleutel SENTIMENT_API_URL=https://uw-sentiment-api/eindpunt SENTIMENT_API_TOKEN=uw_sentiment_token ``` ## 3. Basisimplementatie ```python import os import requests from typing import Dict, Any, List from langchain.agents import AgentType, initialize_agent from langchain.tools import Tool from langchain.llms import OpenAI from langchain.chat_models import ChatOpenAI from langchain.schema import SystemMessage from dotenv import load_dotenv load_dotenv() # OpenWeatherMap Tool class OpenWeatherMapTool: def __init__(self): self.base_url = "https://api.openweathermap.org/data/2.5" self.api_key = os.getenv("OPENWEATHER_API_KEY") def get_weather(self, location: str) -> Dict[str, Any]: """Haal weerinformatie op voor een specifieke locatie""" url = f"{self.base_url}/weather" params = { 'q': location, 'appid': self.api_key, 'units': 'metric', 'lang': 'nl' } try: response = requests.get(url, params=params) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": f"Weer API fout: {str(e)}"} # Aangepaste Sentiment Analyse Tool class SentimentAnalysisTool: def __init__(self): self.api_url = os.getenv("SENTIMENT_API_URL") self.api_token = os.getenv("SENTIMENT_API_TOKEN") def analyze_sentiment(self, text: str) -> Dict[str, Any]: """Analyseer sentiment van tekst met OAuth2 authenticatie""" headers = { 'Authorization': f'Bearer {self.api_token}', 'Content-Type': 'application/json' } payload = { 'text': text, 'language': 'nl' } try: response = requests.post( self.api_url, json=payload, headers=headers, timeout=10 ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: return {"error": f"Sentiment API fout: {str(e)}"} # Rate Limiter voor API-beperkingen class RateLimiter: def __init__(self, max_requests_per_hour: int = 1000): self.request_count = 0 self.max_requests = max_requests_per_hour def can_make_request(self) -> bool: return self.request_count < self.max_requests def increment_count(self): self.request_count += 1 # Dynamische Prompt Bouwer class DynamicPromptBuilder: @staticmethod def build_weather_prompt(weather_data: Dict) -> str: if 'error' in weather_data: return f"Weerdata niet beschikbaar: {weather_data['error']}" return f""" Weer in {weather_data['name']}: Temperatuur: {weather_data['main']['temp']}°C Beschrijving: {weather_data['weather'][0]['description']} Vochtigheid: {weather_data['main']['humidity']}% Wind: {weather_data['wind']['speed']} m/s """ @staticmethod def build_sentiment_prompt(sentiment_data: Dict) -> str: if 'error' in sentiment_data: return f"Sentiment analyse mislukt: {sentiment_data['error']}" return f""" Sentiment Analyse Resultaat: Score: {sentiment_data.get('score', 'N/A')} Emotie: {sentiment_data.get('emotion', 'N/A')} Vertrouwensscore: {sentiment_data.get('confidence', 'N/A')} """ # Main Agent Initialisatie def create_agent(): # Initializeer tools weather_tool = OpenWeatherMapTool() sentiment_tool = SentimentAnalysisTool() rate_limiter = RateLimiter(max_requests_per_hour=1000) # LangChain tools definiëren tools = [ Tool( name="get_weather", func=lambda location: weather_tool.get_weather(location), description="Haal weerinformatie op voor een locatie. Input moet een plaatsnaam zijn." ), Tool( name="analyze_sentiment", func=lambda text: sentiment_tool.analyze_sentiment(text), description="Analyseer sentiment van tekst. Input moet de te analyseren tekst zijn." ) ] # LLM initialiseren llm = ChatOpenAI( temperature=0, model_name="gpt-3.5-turbo", openai_api_key=os.getenv("OPENAI_API_KEY") ) # Systeem prompt voor context system_message = SystemMessage(content=""" Je bent een Nederlandse assistent gespecialiseerd in weeranalyse en sentiment detectie. Gebruik de beschikbare tools om accurate informatie op te halen. Houd rekening met API-beperkingen (max 1000 requests per uur). Geef altijd duidelijke, Nederlandstalige antwoorden. """) # Agent initialiseren agent = initialize_agent( tools=tools, llm=llm, agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verbose=True, agent_kwargs={ 'system_message': system_message } ) return agent, rate_limiter # Voorbeeld gebruik def main(): agent, rate_limiter = create_agent() # Voorbeeld interactie queries = [ "Wat is het weer in Amsterdam?", "Analyseer het sentiment van: 'Ik hou van mooi weer, het maakt me blij'", "Combineer weer in Rotterdam met sentiment analyse van 'Regenachtig weer vind ik niet prettig'" ] for query in queries: if rate_limiter.can_make_request(): try: response = agent.run(query) print(f"Vraag: {query}") print(f"Antwoord: {response}\n") rate_limiter.increment_count() except Exception as e: print(f"Fout bij verwerken vraag: {str(e)}") else: print("Rate limit bereikt. Wacht even voordat je nieuwe requests doet.") break if __name__ == "__main__": main() ``` ## 4. Geavanceerde Workflow Integratie ```python class AdvancedWorkflowManager: def __init__(self, agent, prompt_builder): self.agent = agent self.prompt_builder = prompt_builder def combined_analysis(self, location: str, text: str) -> str: """Combineer weer en sentiment analyse""" weather_data = self.agent.tools[0].func(location) sentiment_data = self.agent.tools[1].func(text) weather_prompt = self.prompt_builder.build_weather_prompt(weather_data) sentiment_prompt = self.prompt_builder.build_sentiment_prompt(sentiment_data) combined_prompt = f""" {weather_prompt} {sentiment_prompt} Geef een geïntegreerde analyse van hoe het weer het sentiment zou kunnen beïnvloeden. """ return self.agent.llm.predict(combined_prompt) # Configuratie voor productie PRODUCTION_CONFIG = { 'max_retries': 3, 'timeout': 30, 'rate_limit_delay': 3.6, # seconden tussen requests voor 1000/uur 'fallback_responses': { 'weather': "Weerinformatie is tijdelijk niet beschikbaar", 'sentiment': "Sentiment analyse service is momenteel offline" } } ``` ## Beperkingen en Overwegingen: 1. **Rate Limiting**: Maximaal 1000 requests per uur - implementeer caching en queueing 2. **Authenticatie**: OAuth2 vereist token refresh mechanisme 3. **Foutafhandeling**: Timeouts en netwerkfouten moeten worden afgevangen 4. **Kostenbeheer**: Monitor API-usage om kosten te controleren 5. **Data Privacy**: Zorg voor encryptie van gevoelige data ## Aanbevolen Verbeteringen: ```python # Voeg caching toe voor veelgebruikte locaties from functools import lru_cache class CachedWeatherTool(OpenWeatherMapTool): @lru_cache(maxsize=100) def get_weather(self, location: str): return super().get_weather(location) # Implementeer retry mechanisme from tenacity import retry, stop_after_attempt, wait_exponential class ResilientSentimentTool(SentimentAnalysisTool): @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def analyze_sentiment(self, text: str): return super().analyze_sentiment(text) ``` Deze implementatie biedt een robuuste basis voor dynamische interactie met beide API's terwijl rekening wordt gehouden met de gespecificeerde beperkingen en authenticatievereisten.