API Error and Agent Flow

(venv) danielkliewer@Daniels-MacBook-Pro Proj % python3 main.py Authenticated as: KonradFreeman Fetched 8 recent posts and comments. --- Iteration 1 --- Processing with ExpandAgent Processing with AnalyzeAgent Processing with MetricAgent HTTP request error: 404 Client Error: Not Found for url: http://localhost:11434/api/generate Process is complete. Final progress has been saved to final_output_20250116_124922.txt. import os import requests import json from utils.base_agent import BaseAgent class FinalAgent(BaseAgent): def __init__(self, model="default-model"): super().__init__() self.endpoint = "http://localhost:11434/api/generate" self.model = model def process(self, prompt): message = prompt.get('message', '') message_str = message if isinstance(message, str) else json.dumps(message, indent=2) data = { "model": self.model, "prompt": f"""Using this analysis: ({message_str}) write a short creative narrative written in the style of the persona of the analysis.""", "stream": False, } try: response = requests.post(self.endpoint, json=data) response.raise_for_status() # Check for HTTP errors json_response = response.json() # Parse JSON response # Get the result from the response result = json_response.get('response', 'No response from API') enhanced_message = f"{message_str}\n\n{result}" return { 'message': enhanced_message, } except requests.exceptions.RequestException as req_err: print(f"HTTP request error: {req_err}") return {'message': message_str} except ValueError as val_err: print(f"Invalid JSON response: {val_err}") return {'message': message_str}import os import requests import json import logging from datetime import datetime import networkx as nx from dotenv import load_dotenv from utils.base_agent import BaseAgent from agents.final_agent import FinalAgent from agents.analyze import AnalyzeAgent from agents.expand import ExpandAgent from utils.reddit_fetch import RedditMonitor from agents.metric_generate import MetricAgent def write_to_file(prompt, filename='output.txt'): with open(filename, 'a') as f: f.write("=== Iteration Output ===\n") f.write("Message:\n") f.write(prompt.get('message', '') + "\n\n") def main(): load_dotenv() # Initialize Reddit monitor reddit_monitor = RedditMonitor() if not reddit_monitor.username: logging.error("Reddit authentication failed. Exiting application.") return reddit_content = reddit_monitor.fetch_all_recent_activity(limit=4) print(f"Fetched {len(reddit_content)} recent posts and comments.") # Initialize agents agents = { 'Expand': ExpandAgent(), 'Analyze': AnalyzeAgent(), 'Metric': MetricAgent(), 'Final': FinalAgent() } # Create a directed graph to model the flow of data between agents G = nx.DiGraph() # Add nodes G.add_nodes_from(agents.keys()) # Define edges to represent the flow between agents G.add_edges_from([ ('Expand', 'Analyze'), ('Analyze', 'Metric'), ('Metric', 'Final'), ('Expand', 'Final'), ('Analyze', 'Final'), ('Metric', 'Final') ]) # Initial prompt prompt = {'message': reddit_content} iteration = 0 max_iterations = 1 # Safety limit to prevent infinite loops is_complete = False while iteration < max_iterations and not is_complete: iteration += 1 print(f"--- Iteration {iteration} ---") # Process the prompt through the agents according to the graph for node in nx.topological_sort(G): if node != 'Final': agent = agents[node] try: print(f"Processing with {node}Agent") prompt = agent.process(prompt) write_to_file(prompt) except Exception as e: logging.error(f"An error occurred in {node}Agent: {e}") return else: # Check completion with the FinalAgent is_complete = agents['Final'].process(prompt) print("Process is complete." if is_complete else "Continuing to next iteration.") if not is_complete: print("Reached maximum iterations without completion. Saving current progress.") # Save the final output timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"final_output_{timestamp}.txt" with open(filename, 'w') as f: f.write(f"Final Output ({timestamp}):\n\n") f.write("Message:\n") f.write(prompt.get('message', '') + "\n\n") print(f"Final progress has been saved to {filename}.") if __name__ == "__main__": main()

Your script demonstrates a modular and orchestrated approach to processing Reddit data using agents represented as nodes in a directed graph. However, there are some points worth noting and improving: ### Issues Identified: 1. **HTTP 404 Error**: - The `FinalAgent` is attempting to send a request to `http://localhost:11434/api/generate`, but the server is not reachable or the endpoint does not exist. This could indicate that the local API server is not running, the port is incorrect, or the endpoint is misconfigured. 2. **Output File Handling**: - The output file (`output.txt` or `final_output_<timestamp>.txt`) is appended or rewritten without robust error handling or checks for disk I/O errors. 3. **Graph Processing Logic**: - While the graph structure is defined and used for agent orchestration, there's room to enhance its flexibility, such as dynamically adjusting flow based on intermediate results. 4. **Logging**: - The logging is minimal. Expanding logging (e.g., using `logging.info` or `logging.debug`) could help in diagnosing issues, especially for HTTP requests and agent processing. 5. **Code Duplication**: - There are several instances where similar operations (e.g., file writing, error handling) are repeated. Refactoring these into helper functions could improve readability and maintainability. --- ### Suggestions for Improvement: 1. **Debugging the 404 Error**: - Verify that the API server is running on the correct port (`11434`) and that the `/api/generate` endpoint is properly configured. - Test the endpoint manually using tools like `curl` or Postman to confirm its availability. 2. **Improving Agent Orchestration**: - Allow the flow of data between agents to adapt dynamically based on conditions (e.g., skipping certain agents if their output is unnecessary). - Example: ```python if some_condition_based_on_metric_analysis: continue # Skip to next relevant agent ``` 3. **Enhanced Logging**: - Integrate more descriptive logs: ```python logging.info(f"Agent {node} processing started.") logging.debug(f"Prompt before {node}: {prompt}") ``` - Write errors to a dedicated log file for debugging: ```python logging.basicConfig(filename='application.log', level=logging.DEBUG) ``` 4. **Error Handling**: - Centralize error handling logic for cleaner code: ```python def safe_process(agent, prompt): try: return agent.process(prompt) except Exception as e: logging.error(f"Error in {agent.__class__.__name__}: {e}") return prompt ``` 5. **Test the API Call Independently**: - Before running the full application, ensure the API call works by testing `FinalAgent` in isolation: ```python final_agent = FinalAgent() test_prompt = {'message': 'Test message'} print(final_agent.process(test_prompt)) ``` 6. **Add More Iterations and Safeguards**: - Instead of a single iteration, allow for more flexible retries, with a safeguard to prevent infinite loops: ```python while iteration < max_iterations and not is_complete: try: # Process iteration except Exception as e: logging.error(f"Error during iteration {iteration}: {e}") break ``` --- ### Next Steps: - Debug the local API server issue to resolve the 404 error. - Refactor the code to improve modularity, logging, and error handling. - Test agents individually to ensure each performs as expected. - Expand the graph logic to accommodate dynamic workflows. Would you like help troubleshooting the API server issue or refactoring the code further?