support batch

Author: zachary62

.cursorrules

@@ -715,7 +715,7 @@ class RetryNode(Node):
 To **gracefully handle** the exception (after all retries) rather than raising it, override:
 
 ```python 
-def exec_fallback(self, shared, prep_res, exc):
+def exec_fallback(self, prep_res, exc):
     raise exc
 ```
 
@@ -735,7 +735,7 @@ class SummarizeFile(Node):
         summary = call_llm(prompt)  # might fail
         return summary
 
-    def exec_fallback(self, shared, prep_res, exc):
+    def exec_fallback(self, prep_res, exc):
         # Provide a simple fallback instead of crashing
         return "There was an error processing your request."
 

flow.py

@@ -57,6 +57,12 @@ def exec(self, url):
         content = get_html_content(url)
         return {"url": url, "content": content}
 
+    def exec_fallback(self, prep_res, exc):
+        # This is called after all retries are exhausted
+        url = prep_res["url"]  # Extract URL from the prep_res input pair
+        logger.error(f"Failed to retrieve content from {url} after all retries: {exc}")
+        return {"url": url, "content": None}
+    
     def post(self, shared, prep_res, exec_res_list):
         # Store only non-empty webpage contents
         valid_contents = [res for res in exec_res_list if res["content"]]
@@ -97,9 +103,11 @@ def exec(self, url_content_pair):
 ```yaml
 factors:
     - name: "factor_name"
+    action: "action to take"
     actionable: true/false
     details: "supporting details if actionable"
     - name: "another_factor"
+    action: "action to take"
     actionable: true/false
     details: "supporting details if actionable"
 ```"""
@@ -115,7 +123,7 @@ def exec(self, url_content_pair):
         logger.debug(f"Successfully parsed YAML from LLM response for {url}")
         return {"url": url, "analysis": analysis}
 
-    def exec_fallback(self, shared, prep_res, exc):
+    def exec_fallback(self, prep_res, exc):
         # This is called after all retries are exhausted
         url = prep_res["url"]  # Extract URL from the prep_res input pair
         logger.error(f"Failed to analyze content from {url} after all retries: {exc}")
@@ -233,8 +241,8 @@ def post(self, shared, prep_res, exec_res):
 logger.info("Initializing flow nodes")
 search_node = SearchPersonNode()
 content_node = ContentRetrievalNode()
-analyze_node = AnalyzeResultsBatchNode(max_retries=3)  # Retry up to 3 times before using fallback
-draft_node = DraftOpeningNode()
+analyze_node = AnalyzeResultsBatchNode(max_retries=2, wait=10)  # Retry up to 3 times before using fallback
+draft_node = DraftOpeningNode(max_retries=3, wait=10)
 
 # Connect nodes in the flow
 logger.info("Connecting nodes in personalization flow")

input.csv

@@ -0,0 +1,4 @@
+first_name,last_name,keywords
+Elon,Musk,Tesla SpaceX entrepreneur
+Sundar,Pichai,Google CEO tech
+Tim,Cook,Apple iPhone CEO 

main_batch.py

@@ -0,0 +1,145 @@
+import csv
+import argparse
+import os
+import json
+from flow import cold_outreach_flow
+
+def main():
+    """
+    Batch processing script for the Cold Outreach Opener Generator.
+    Processes multiple people from a CSV file and generates personalized opening messages.
+    """
+    # Parse command line arguments
+    parser = argparse.ArgumentParser(description='Process multiple cold outreach targets from a CSV file.')
+    parser.add_argument('--input', default='input.csv', help='Input CSV file (default: input.csv)')
+    parser.add_argument('--output', default='output.csv', help='Output CSV file (default: output.csv)')
+    args = parser.parse_args()
+    
+    # Check if input file exists
+    if not os.path.exists(args.input):
+        print(f"Error: Input file '{args.input}' not found.")
+        return
+    
+    # Hardcoded personalization factors
+    personalization_factors = [
+        {
+            "name": "personal_connection",
+            "description": "Check if target person has Columbia University affiliation",
+            "action": "If they do, mention shared connection to Columbia"
+        },
+        {
+            "name": "recent_promotion",
+            "description": "Check if target person was recently promoted",
+            "action": "If they were, congratulate them on their new role"
+        },
+        {
+            "name": "recent_talks",
+            "description": "Check if target person gave talks recently",
+            "action": "If they did, mention enjoying their insights"
+        }
+    ]
+    
+    # Extract factor names for later use
+    factor_names = [factor["name"] for factor in personalization_factors]
+    print(factor_names)
+    
+    # Hardcoded style preference
+    style = "Be concise, specific, and casual in 30 words or less. For example: 'Heard about your talk on the future of space exploration—loved your take on creating a more sustainable path for space travel.'"
+    
+    # Read input CSV
+    input_data = []
+    with open(args.input, 'r', newline='', encoding='utf-8') as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            if 'first_name' in row and 'last_name' in row and 'keywords' in row:
+                input_data.append(row)
+    
+    if not input_data:
+        print(f"Error: No valid data found in '{args.input}'. CSV should have columns: first_name, last_name, keywords")
+        return
+    
+    # Process each person
+    results = []
+    total = len(input_data)
+    for i, person in enumerate(input_data, 1):
+        print(f"\nProcessing {i}/{total}: {person['first_name']} {person['last_name']}")
+        
+        # Prepare input data
+        shared = {
+            "input": {
+                "first_name": person['first_name'],
+                "last_name": person['last_name'],
+                "keywords": person['keywords'],
+                "personalization_factors": personalization_factors,
+                "style": style
+            }
+        }
+        
+        # Run the flow
+        try:
+            cold_outreach_flow.run(shared)
+            
+            # Prepare result row
+            # Extract URLs as comma-separated string
+            urls = [result.get("link", "") for result in shared.get("search_results", []) if "link" in result]
+            url_string = ",".join(urls)
+            
+            # Extract personalization details
+            personalization_data = {}
+            for factor_name, details in shared.get("personalization", {}).items():
+                personalization_data[factor_name + "_actionable"] = str(details.get("actionable", False))
+                personalization_data[factor_name + "_details"] = details.get("details", "")
+            
+            result = {
+                'first_name': person['first_name'],
+                'last_name': person['last_name'],
+                'keywords': person['keywords'],
+                'opening_message': shared.get("output", {}).get("opening_message", ""),
+                'search_results': url_string,
+                **personalization_data  # Add all personalization fields
+            }
+            results.append(result)
+            
+            # Display the result
+            print(f"Generated opener: {result['opening_message']}")
+            
+        except Exception as e:
+            print(f"Error processing {person['first_name']} {person['last_name']}: {str(e)}")
+            # Add failed row with error message
+            results.append({
+                'first_name': person['first_name'],
+                'last_name': person['last_name'],
+                'keywords': person['keywords'],
+                'opening_message': f"ERROR: {str(e)}",
+                'search_results': "",
+                # Include empty personalization fields for consistency with successful rows
+                **{f"{factor}_actionable": "False" for factor in factor_names},
+                **{f"{factor}_details": "" for factor in factor_names}
+            })
+    
+    # Write results to output CSV
+    if results:
+        # Determine all field names by examining all the result rows
+        all_fields = set()
+        for result in results:
+            all_fields.update(result.keys())
+        
+        fieldnames = ['first_name', 'last_name', 'keywords', 'opening_message', 'search_results']
+        # Add personalization fields in a specific order
+        for factor in factor_names:
+            if f"{factor}_actionable" in all_fields:
+                fieldnames.append(f"{factor}_actionable")
+            if f"{factor}_details" in all_fields:
+                fieldnames.append(f"{factor}_details")
+        
+        with open(args.output, 'w', newline='', encoding='utf-8') as csvfile:
+            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
+            writer.writeheader()
+            writer.writerows(results)
+        
+        print(f"\nProcessing complete. Results written to '{args.output}'")
+    else:
+        print("\nNo results to write.")
+
+if __name__ == "__main__":
+    main() 

output.csv

@@ -0,0 +1,6 @@
+first_name,last_name,keywords,opening_message,search_results,recent_promotion_actionable,recent_promotion_details,recent_talks_actionable,recent_talks_details
+Elon,Musk,Tesla SpaceX entrepreneur,"Elon, congratulations on heading DOGE! Caught your recent Rogan interview—your insights on streamlining government efficiency while balancing innovation priorities were fascinating.","https://www.tesla.com/elon-musk,https://en.wikipedia.org/wiki/Elon_Musk,https://www.investopedia.com/articles/personal-finance/061015/how-elon-musk-became-elon-musk.asp,https://x.com/elonmusk?lang=en,https://www.biography.com/business-leaders/elon-musk,https://www.reddit.com/r/IAmA/comments/kkn5v/ask_tesla_spacex_and_paypal_cofounder_elon_musk/,https://twitter.com/elonmusk,https://www.reddit.com/r/SpaceXMasterrace/comments/sasbcw/elon_musk_was_not_the_founder_of_spacex_he_took/,https://www.space.com/18849-elon-musk.html,https://www.forbes.com/profile/elon-musk/",True,"Elon Musk was appointed as a senior advisor to United States president Donald Trump and de facto head of the Department of Government Efficiency (DOGE) in January 2025. | The content mentions Elon Musk's very recent appointment as (unofficial) head of the so-called Department of Government Efficiency (DOGE) following the 2024 election, where he now has 'unprecedented sway over federal policy and spending.' This appears to be a significant new role for him in government. | Elon Musk was recently appointed to lead the Department of Government Efficiency (DOGE) under President Trump's second administration in January 2025. This is a temporary organization tasked with cutting unnecessary federal spending and modernizing technology within the federal government. | Elon Musk recently became part of the government as head of the Department of Government Efficiency (DOGE). The content references his 'new government role' and discusses his activities leading DOGE, including sending emails to federal employees.",True,"Musk recently spoke at Trump's Presidential Parade in January 2025, where he made controversial remarks and gestures. He also gave an interview with psychologist Jordan Peterson in 2024 where he discussed his religious views, stating he considers himself a 'cultural Christian' and believes in the teachings of Jesus Christ. | Musk was recently interviewed by Joe Rogan (mentioned as 'Joe Rogan Interviews Elon Musk: Here Are The Key Moments' from February 28, 2025) and also spoke at the Conservative Political Action Conference on February 20, 2025."
+Sundar,Pichai,Google CEO tech,"Hi Sundar,
+
+Your recent lecture at Carnegie Mellon about AI becoming a ""natural extension"" of daily life really resonated with me, especially your insights on Project Astra and DeepMind's mathematical breakthroughs.","https://en.wikipedia.org/wiki/Sundar_Pichai,https://www.linkedin.com/in/sundarpichai,https://mitsloan.mit.edu/ideas-made-to-matter/googles-sundar-pichai-says-tech-a-powerful-agent-change,https://www.productplan.com/learn/sundar-pichai-product-manager-to-ceo/,https://blog.google/authors/sundar-pichai/,https://www.weforum.org/stories/2020/01/this-is-how-quantum-computing-will-change-our-lives-8a0d33657f/,https://www.cmu.edu/news/stories/archives/2024/september/google-ceo-sundar-pichai-launches-the-2024-25-presidents-lecture-series-at-carnegie-mellon,https://www.nytimes.com/2018/11/08/business/sundar-pichai-google-corner-office.html,https://www.youtube.com/watch?v=OsxwBmp3iFU,https://www.theguardian.com/technology/2015/aug/10/google-sundar-pichai-ceo-alphabet",True,"The article announces Sundar Pichai's promotion to CEO of Google as part of a restructuring where Google became a subsidiary of Alphabet, with Larry Page moving to head Alphabet. This was announced on August 10, 2015.",True,"The content describes Pichai giving a talk at MIT Forefront where he discussed AI, future of work, diversity and inclusion, and climate change with MIT president L. Rafael Reif in March 2022 | The content mentions 'Google I/O 2024' where Sundar Pichai shared news about updates across Gemini, Android, Search and Photos, which indicates he recently gave a presentation or talk at this event. | Sundar Pichai recently gave a lecture at Carnegie Mellon University on September 18, 2024, titled 'The AI Platform Shift and the Opportunity Ahead' as part of the President's Lecture Series. He discussed AI innovations including Project Astra, Google DeepMind's success in the International Mathematical Olympiad, and his vision for AI becoming a 'natural extension' in everyday life. | The webpage is about a talk/interview titled 'Building the Future: Sundar Pichai on A.I., Regulation and What's Next for Google' which appears to be a recent discussion where he shared insights about AI, regulation, and Google's future."
+Tim,Cook,Apple iPhone CEO ,"Tim, your recent London interview about your Apple journey was refreshing. I especially appreciated your candid reflections on how Steve Jobs recruited you—it offers valuable perspective for today's tech leaders.","https://en.wikipedia.org/wiki/Tim_Cook,https://www.apple.com/leadership/tim-cook/,https://www.youtube.com/watch?v=m4RVTK7iU1c,https://www.apple.com/leadership/,https://x.com/tim_cook/status/1890068457825394918?lang=en,https://stratechery.com/2013/tim-cook-is-a-great-ceo/,https://x.com/tim_cook?lang=en,https://www.reddit.com/r/apple/comments/xa0vg9/tim_cook_has_been_ceo_of_apple_since_2011_since/,https://investor.apple.com/our_values/default.aspx,https://www.forbes.com/profile/tim-cook/",,,True,"The webpage shows Tim Cook giving an interview titled 'The Job Interview' where he discusses how Steve Jobs recruited him and other topics. This indicates a recent talk/interview that could be referenced. | Tim Cook was interviewed in London recently (mentioned in Dec 14, 2024 article: 'Apple CEO Tim Cook Declares New iPhone Era In Detailed Interview'). The interview covered business advice, his personal story, and Apple Intelligence."