simplify prompt

assets/prompt

@@ -1,10 +1,8 @@
-# Example App for text summarization & QA using minillmflow
+from minillmflow import *
-from minillmflow import Node, BatchNode, Flow, BatchFlow, AsyncNode, AsyncFlow, BatchAsyncFlow
+import openai, os, yaml
-import os
 
-# 1) Implement a simple LLM helper (OpenAI in this example).
+# Minimal LLM wrapper
 def call_llm(prompt):
-    # Users must set an OpenAI API key; can also load from env var, etc.
     openai.api_key = "YOUR_API_KEY_HERE"
     r = openai.ChatCompletion.create(
         model="gpt-4",
@@ -12,211 +10,113 @@ def call_llm(prompt):
     )
     return r.choices[0].message.content
 
-# 2) Create a shared store (dict) for Node/Flow data exchange.
-#    This can be replaced with a DB or other storage.
-#    Design the structure / schema based on the app requirements.
 shared = {"data": {}, "summary": {}}
 
-# 3) Create a Node that loads data from disk into shared['data'].
+# Load data into shared['data']
 class LoadData(Node):
-    # For compute-intensive operations, do them in prep().
     def prep(self, shared):
         path = "../data/PaulGrahamEssaysLarge"
-        for filename in os.listdir(path):
+        for fn in os.listdir(path):
-            with open(os.path.join(path, filename), 'r') as f:
+            with open(os.path.join(path, fn), 'r') as f:
-                shared['data'][filename] = f.read()
+                shared['data'][fn] = f.read()
-    # If LLM was needed, we'd handle it in exec(). Not needed here.
+    def exec(self, res): pass
-    # (idempotent so it can be retried if needed)
+    def post(self, s, pr, er): pass
-    def exec(self,shared,prep_res): pass 
-    # post() can update shared again or decide the next node (by return the action).
-    def post(self,shared,prep_res,exec_res): pass 
 
-load_data = LoadData()
+LoadData().run(shared)
-# Run the data-loading node once
-load_data.run(shared)
 
-# 4) Create a Node that summarizes a single file using the LLM.
+# Summarize one file
 class SummarizeFile(Node):
-    def prep(self, shared):
+    def prep(self, s): return s['data'][self.params['filename']]
-        # Use self.params (which must remain immutable during prep/exec/post).
+    def exec(self, content):
-        # Typically, we only store identifying info in params (e.g., filename).
+        return call_llm(f"{content} Summarize in 10 words.")
-        content = shared['data'][self.params['filename']]
+    def post(self, s, pr, sr): s["summary"][self.params['filename']] = sr
-        return content
-    def exec(self, shared, prep_res):
-        content = prep_res
-        prompt = f"{content} Respond a summary of above in 10 words"
-        summary = call_llm(prompt)
-        return summary
-    def post(self, shared, prep_res, exec_res):
-        shared["summary"][self.params['filename']] = exec_res
 
-summarize_file = SummarizeFile()
+node_summ = SummarizeFile()
-# For testing, we set params directly on the node.
+node_summ.set_params({"filename":"addiction.txt"})
-# In real usage, you'd set them in a Flow or BatchFlow.
+node_summ.run(shared)
-summarize_file.set_params({"filename":"addiction.txt"})
-summarize_file.run(shared)
 
-# 5) If data is large, we can apply a map-reduce pattern:
+# Map-Reduce summarization
-#    - MapSummaries(BatchNode) => chunk the file and summarize each chunk
-#    - ReduceSummaries(Node)   => combine those chunk-level summaries
 class MapSummaries(BatchNode):
-    def prep(self, shared):
+    def prep(self, s):
-        content = shared['data'][self.params['filename']]
+        text = s['data'][self.params['filename']]
-        chunk_size = 10000
+        return [text[i:i+10000] for i in range(0, len(text), 10000)]
-        chunks = [content[i:i+chunk_size] for i in range(0, len(content), chunk_size)]
+    def exec(self, chunk):
-        # Must return an iterable (list or generator) for a BatchNode.
+        return call_llm(f"{chunk} Summarize in 10 words.")
-        return chunks
+    def post(self, s, pr, er):
-    def exec(self, shared, prep_res):
+        s["summary"][self.params['filename']] = [f"{i}. {r}" for i,r in enumerate(er)]
-        # Each iteration of prep_res corresponds to a single chunk.
-        chunk = prep_res
-        prompt = f"{chunk} Respond a summary of above in 10 words"
-        summary = call_llm(prompt)
-        return summary
-    def post(self, shared, prep_res, exec_res):
-        # exec_res is a list of exec() results (summaries for each chunk).
-        combined_summary = [f"{i}. {summary}" for i, summary in enumerate(exec_res)]
-        shared["summary"][self.params['filename']] = combined_summary
 
 class ReduceSummaries(Node):
-    def prep(self, shared):
+    def prep(self, s): return s["summary"][self.params['filename']]
-        # Retrieve the list of chunk summaries from shared storage
+    def exec(self, chunks):
-        return shared["summary"][self.params['filename']]
+        return call_llm(f"{chunks} Combine into 10 words summary.")
-    def exec(self, shared, prep_res):
+    def post(self, s, pr, sr): s["summary"][self.params['filename']] = sr
-        combined_summary = prep_res
-        prompt = f"{combined_summary} Respond a summary of above in 10 words"
-        summary = call_llm(prompt)
-        return summary
-    def post(self, shared, prep_res, exec_res):
-        # Store the combined summary as the final summary for this file.
-        shared["summary"][self.params['filename']] = exec_res
-        
-map_summaries = MapSummaries()
-reduce_summaries = ReduceSummaries()
-# Link map_summaries to reduce_summaries with an action
-# By default, the action is "default" (when post returns None, it takes "default" action)
-# This is the same as map_summaries - "default" >> reduce_summaries
-map_summaries >> reduce_summaries
 
-# We don't directly call map_summaries.run(shared), 
+map_summ = MapSummaries()
-# because that alone would process only the map step without reduce.
+reduce_summ = ReduceSummaries()
+map_summ >> reduce_summ
 
-# 6) Instead, create a Flow that starts from map_summaries (a Node) 
+flow = Flow(start=map_summ)
-#    and automatically includes reduce_summaries. 
+flow.set_params({"filename":"before.txt"})
-#    Note: A Flow can also start from any other Flow or BatchFlow.
+flow.run(shared)
 
-
+# Summarize all files
-file_summary_flow = Flow(start=map_summaries)
-# When a flow params is set, it will recursively set its params to all nodes in the flow
-file_summary_flow.set_params({"filename":"before.txt"})
-file_summary_flow.run(shared)
-
-# 7) Summarize all files using a BatchFlow that reruns file_summary_flow for each file
 class SummarizeAllFiles(BatchFlow):
-    def prep(self, shared):
+    def prep(self, s): return [{"filename":fn} for fn in s['data']]
-        # Return a list of parameters to apply in each flow iteration.
-        # Each individual param will be merged with this node's own params 
-        # Allowing nesting of multi-level BatchFlow. 
-        # E.g., first level diretcory, second level file.
-        return [{"filename":filename} for filename in shared['data']]
 
-summarize_all_files = SummarizeAllFiles(start=file_summary_flow)
+SummarizeAllFiles(start=flow).run(shared)
-summarize_all_files.run(shared)
 
-
+# QA agent
-# 8) QA Agent: Find the most relevant file based on summary with actions
-#    if no question is asked:
-#       (a) end: terminate the flow 
-#    if question is asked:
-#         if relevant file is found:
-#            (b) answer: move to answer node and read the whole file to answer the question
-#         if no relevant file is found:
-#            (c) retry: retry the process to find the relevant file
 class FindRelevantFile(Node):
-    def prep(self, shared):
+    def prep(self, s):
-        question = input("Enter a question: ")
+        q = input("Enter a question: ")
-        formatted_list = [f"- '{filename}': {shared['summary'][filename]}" 
+        summ = [f"- '{fn}': {s['summary'][fn]}" for fn in s['summary']]
-                         for filename in shared['summary']]
+        return q, summ
-        return question, formatted_list
+    def exec(self, p):
-    def exec(self, shared, prep_res):
+        q, summ = p
-        question, formatted_list = prep_res
+        if not q:
-        if not question:
+            return {"think":"no question","has_relevant":False}
-            return {"think":"no question", "has_relevant":False}
+        resp = call_llm(f"""
-        # Provide a structured YAML output that includes:
+Question: {q} 
-        # - The chain of thought
+Find the most relevant file from: {summ}
-        # - Whether any relevant file was found
+If none, explain why
-        # - The most relevant file if found
+Respond in YAML:
-        prompt = f"""Question: {question} 
+think: ...
-Find the most relevant file from: 
+has_relevant: ...
-{formatted_list}
+most_relevant: ...
-If no relevant file, explain why
+""")
-Respond in yaml without additional information:
+        r = yaml.safe_load(resp)
-think: the question has/has no relevant file ...
+        return r
-has_relevant: true/false
+    def exec_fallback(self, p, exc): return {"think":"error","has_relevant":False}
-most_relevant: filename"""
+    def post(self, s, pr, res):
-        response = call_llm(prompt)
+        q, _ = pr
-        import yaml
+        if not q:
-        result = yaml.safe_load(response)
+            print("No question asked"); return "end"
-        # Ensure required fields are present
+        if res["has_relevant"]:
-        assert "think" in result
+            s["question"], s["relevant_file"] = q, res["most_relevant"]
-        assert "has_relevant" in result
+            print("Relevant file:", res["most_relevant"])
-        assert "most_relevant" in result if result["has_relevant"] else True
-        return result
-    # handle errors by returning a default response in case of exception after retries
-    def exec_fallback(self,shared,prep_res,exc):
-        # if not overridden, the default is to throw the exception
-        return {"think":"error finding the file", "has_relevant":False}
-    def post(self, shared, prep_res, exec_res):
-        question, _ = prep_res
-        # Decide what to do next based on the results
-        if not question:
-            print(f"No question asked")
-            return "end"
-        if exec_res["has_relevant"]:
-            # Store the question and most relevant file in shared
-            shared["question"] = question
-            shared["relevant_file"] = exec_res['most_relevant']
-            print(f"Relevant file found: {exec_res['most_relevant']}")
             return "answer"
         else:
-            print(f"No relevant file found: {exec_res['think']}")
+            print("No relevant file:", res["think"])
             return "retry"
 
 class AnswerQuestion(Node):
-    def prep(self, shared):
+    def prep(self, s):
-        question = shared['question']
+        return s['question'], s['data'][s['relevant_file']]
-        relevant_file = shared['relevant_file']
+    def exec(self, p):
-        # Read the whole file content
+        q, txt = p
-        file_content = shared['data'][relevant_file]
+        return call_llm(f"Question: {q}\nText: {txt}\nAnswer in 50 words.")
-        return question, file_content
+    def post(self, s, pr, ex):
-    def exec(self, shared, prep_res):
+        print("Answer:", ex)
-        question, file_content = prep_res
-        prompt = f"""Question: {question}
-File: {file_content}
-Answer the question in 50 words"""
-        response = call_llm(prompt)
-        return response
-    def post(self, shared, prep_res, exec_res):
-        print(f"Answer: {exec_res}")
 
-class NoOp(Node):
+class NoOp(Node): pass
-    pass
 
-# Configure the QA agent with appropriate transitions and retries
+frf = FindRelevantFile(max_retries=3)
-find_relevant_file = FindRelevantFile(max_retries=3)
+aq = AnswerQuestion()
-answer_question = AnswerQuestion()
+noop = NoOp()
-no_op = NoOp()
 
-# Connect the nodes based on the actions they return
+frf - "answer" >> aq >> frf
-find_relevant_file - "answer" >> answer_question >> find_relevant_file
+frf - "retry"  >> frf
-find_relevant_file - "retry" >> find_relevant_file
+frf - "end"    >> noop
-find_relevant_file - "end" >> no_op
 
-qa_agent = Flow(start=find_relevant_file)
+qa = Flow(start=frf)
-qa_agent.run(shared)
+qa.run(shared)
-
-
-# Above example demonstrates the use of minillmflow
-# Next, build another app based on the same principles
-# First, given the app's requirements, design the Node/Flow structure
-# Then, design the data structure within shared storage, and how it's updated
-# Finally, implement the Nodes and Flows to achieve the desired functionality

cookbook/demo.ipynb

@@ -355,7 +355,7 @@
         "        assert \"most_relevant\" in result if result[\"has_relevant\"] else True\n",
         "        return result\n",
         "    # handle errors by returning a default response in case of exception after retries\n",
-        "    def exec_fallback(self,shared,prep_res,exc):\n",
+        "    def exec_fallback(self,prep_res,exc):\n",
         "        # if not overridden, the default is to throw the exception\n",
         "        return {\"think\":\"error finding the file\", \"has_relevant\":False}\n",
         "    def post(self, shared, prep_res, exec_res):\n",