add TAO example

relic-yuexi · relic-yuexi · commit ad16fb701318 · 2025-06-06T11:02:35.000+08:00
diff --git a/cookbook/pocketflow-tao/README.md b/cookbook/pocketflow-tao/README.md
@@ -0,0 +1,103 @@
+# PocketFlow TAO (Thought-Action-Observation)
+
+A powerful pattern that enables AI agents to solve complex problems through structured thinking, action execution, and result observation. This example demonstrates how to implement the TAO pattern using PocketFlow.
+
+## Project Structure
+
+```
+.
+├── flow.py        # PocketFlow implementation of TAO pattern
+├── main.py        # Main application entry point
+├── nodes.py       # TAO node definitions
+├── requirements.txt # Project dependencies
+└── README.md      # Project documentation
+```
+
+## Overview
+
+The TAO pattern consists of three key steps:
+1. **Thought**: The agent deeply analyzes the problem and forms a solution strategy
+2. **Action**: Concrete actions are executed based on the thinking
+3. **Observation**: Results are evaluated and feedback is gathered
+
+This cycle continues until the problem is solved or termination conditions are met.
+
+## Setup
+
+1. Create a virtual environment:
+```bash
+python -m venv venv
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+```
+
+2. Install dependencies:
+```bash
+pip install -r requirements.txt
+```
+
+3. Set API key (if using specific LLM services):
+```bash
+export OPENAI_API_KEY="your-api-key-here"
+# Or set in code
+```
+
+## How to Run
+
+Execute the example:
+```bash
+python main.py
+```
+
+## How It Works
+
+The TAO pattern is implemented as a flow in PocketFlow, with each step handled by specialized nodes:
+
+```mermaid
+graph TD
+    Problem[Problem Input] --> ThoughtNode
+    ThoughtNode[Thought Node] --> ActionNode[Action Node]
+    ActionNode --> ObservationNode[Observation Node]
+    ObservationNode --> DecisionNode{Problem Solved?}
+    DecisionNode -->|Yes| Solution[Solution]
+    DecisionNode -->|No| ThoughtNode
+```
+
+Each TAO cycle generates new insights for the problem-solving process, allowing the AI to iteratively approach an optimal solution.
+
+## Use Cases
+
+- Complex problem solving
+- Multi-step reasoning tasks
+- Projects requiring iterative improvement
+- Reinforcement learning-style AI applications
+
+## Example Output
+
+```
+Query: I need to understand the latest developments in artificial intelligence
+
+🤔 Thought 1: Decided to execute search
+🚀 Executing action: search, input: latest developments in artificial intelligence 2023
+✅ Action completed, result obtained
+👁️ Observation: The search result indicates that information was r...
+🎯 Final Answer: As of October 2023, some of the latest developments in artificial intelligence include advances in large language models like GPT-4, increased focus on AI alignment and safety, improvements in reinforcement learning, and the integration of AI into more industries such as healthcare, finance, and autonomous vehicles. Researchers are also exploring ethical considerations and regulatory frameworks to ensure responsible AI deployment. For the most current updates beyond this date, I recommend checking recent publications, official AI research organization releases, or news sources specializing in technology.
+
+Flow ended, thank you for using!
+
+Final Answer:
+As of October 2023, some of the latest developments in artificial intelligence include advances in large language models like GPT-4, increased focus on AI alignment and safety, improvements in reinforcement learning, and the integration of AI into more industries such as healthcare, finance, and autonomous vehicles. Researchers are also exploring ethical considerations and regulatory frameworks to ensure responsible AI deployment. For the most current updates beyond this date, I recommend checking recent publications, official AI research organization releases, or news sources specializing in technology.
+```
+
+## Advanced Usage
+
+The TAO pattern can be extended by:
+- Adding memory components to store past thoughts and observations.
+- Implementing adaptive action selection strategies.
+- Integrating external tools and APIs.
+- Adding human feedback loops.
+- Adding max attempt to control the iteration.
+
+## Additional Resources
+
+- [PocketFlow Documentation](https://the-pocket.github.io/PocketFlow/)
+- [Understanding AI Agents through the Thought-Action-Observation Cycle](https://huggingface.co/learn/agents-course/en/unit1/agent-steps-and-structure)
diff --git a/cookbook/pocketflow-tao/flow.py b/cookbook/pocketflow-tao/flow.py
@@ -0,0 +1,39 @@
+# flow.py
+
+from pocketflow import Flow
+from nodes import ThinkNode, ActionNode, ObserveNode, EndNode
+
+def create_tao_flow():
+    """
+    Create a Thought-Action-Observation loop flow
+    
+    How the flow works:
+    1. ThinkNode decides the next action
+    2. ActionNode executes the action
+    3. ObserveNode observes the action result
+    4. Return to ThinkNode to continue thinking, or end the flow
+    
+    Returns:
+        Flow: Complete TAO loop flow
+    """
+    # Create node instances
+    think = ThinkNode()
+    action = ActionNode()
+    observe = ObserveNode()
+    end = EndNode()
+    
+    # Connect nodes
+    # If ThinkNode returns "action", go to ActionNode
+    think - "action" >> action
+    
+    # If ThinkNode returns "end", end the flow
+    think - "end" >> end
+    
+    # After ActionNode completes, go to ObserveNode
+    action - "observe" >> observe
+    
+    # After ObserveNode completes, return to ThinkNode
+    observe - "think" >> think
+    
+    # Create and return flow, starting from ThinkNode
+    return Flow(start=think)
diff --git a/cookbook/pocketflow-tao/main.py b/cookbook/pocketflow-tao/main.py
@@ -0,0 +1,29 @@
+# main.py
+
+from flow import create_tao_flow
+
+def main():
+    
+    query = """I need to understand the latest developments in artificial intelligence"""
+    
+    # Create shared data
+    shared = {
+        "query": query,
+        "thoughts": [],
+        "observations": [],
+        "current_thought_number": 0
+    }
+    
+    # Create and run flow
+    tao_flow = create_tao_flow()
+    tao_flow.run(shared)
+    
+    # Print final result
+    if "final_answer" in shared:
+        print("\nFinal Answer:")
+        print(shared["final_answer"])
+    else:
+        print("\nFlow did not produce a final answer")
+
+if __name__ == "__main__":
+    main()
diff --git a/cookbook/pocketflow-tao/nodes.py b/cookbook/pocketflow-tao/nodes.py
@@ -0,0 +1,190 @@
+# nodes.py
+
+from pocketflow import Node
+import yaml
+from utils import call_llm
+
+class ThinkNode(Node):
+    def prep(self, shared):
+        """Prepare the context needed for thinking"""
+        query = shared.get("query", "")
+        observations = shared.get("observations", [])
+        thoughts = shared.get("thoughts", [])
+        current_thought_number = shared.get("current_thought_number", 0)
+        
+        # Update thought count
+        shared["current_thought_number"] = current_thought_number + 1
+        
+        # Format previous observations
+        observations_text = "\n".join([f"Observation {i+1}: {obs}" for i, obs in enumerate(observations)])
+        if not observations_text:
+            observations_text = "No observations yet."
+            
+        return {
+            "query": query,
+            "observations_text": observations_text,
+            "thoughts": thoughts,
+            "current_thought_number": current_thought_number + 1
+        }
+    
+    def exec(self, prep_res):
+        """Execute the thinking process, decide the next action"""
+        query = prep_res["query"]
+        observations_text = prep_res["observations_text"]
+        current_thought_number = prep_res["current_thought_number"]
+        
+        # Build the prompt
+        prompt = f"""
+        You are an AI assistant solving a problem. Based on the user's query and previous observations, think about what action to take next.
+        
+        User query: {query}
+        
+        Previous observations:
+        {observations_text}
+        
+        Please think about the next action and return your thinking process and decision in YAML format:
+        ```yaml
+        thinking: |
+            <detailed thinking process>
+        action: <action name, such as 'search' or 'answer'>
+        action_input: <input parameters for the action>
+        is_final: <set to true if this is the final answer, otherwise false>
+        ```
+        """
+        
+        # Call LLM to get thinking result
+        response = call_llm(prompt)
+        
+        # Parse YAML response
+        yaml_str = response.split("```yaml")[1].split("```")[0].strip()
+        thought_data = yaml.safe_load(yaml_str)
+        
+        # Add thought number
+        thought_data["thought_number"] = current_thought_number
+        
+        return thought_data
+    
+    def post(self, shared, prep_res, exec_res):
+        """Save the thinking result and decide the next step in the flow"""
+        # Save thinking result
+        if "thoughts" not in shared:
+            shared["thoughts"] = []
+        shared["thoughts"].append(exec_res)
+        
+        # Save action information
+        shared["current_action"] = exec_res["action"]
+        shared["current_action_input"] = exec_res["action_input"]
+        
+        # If it's the final answer, end the flow
+        if exec_res.get("is_final", False):
+            shared["final_answer"] = exec_res["action_input"]
+            print(f"🎯 Final Answer: {exec_res['action_input']}")
+            return "end"
+        
+        # Otherwise continue with the action
+        print(f"🤔 Thought {exec_res['thought_number']}: Decided to execute {exec_res['action']}")
+        return "action"
+
+class ActionNode(Node):
+    def prep(self, shared):
+        """Prepare to execute action"""
+        action = shared["current_action"]
+        action_input = shared["current_action_input"]
+        return action, action_input
+    
+    def exec(self, inputs):
+        """Execute action and return result"""
+        action, action_input = inputs
+        
+        print(f"🚀 Executing action: {action}, input: {action_input}")
+        
+        # Execute different operations based on action type
+        if action == "search":
+            # Simulate search operation
+            result = self.search_web(action_input)
+        elif action == "calculate":
+            # Simulate calculation operation
+            result = self.calculate(action_input)
+        elif action == "answer":
+            # Direct return answer
+            result = action_input
+        else:
+            # Unknown action type
+            result = f"Unknown action type: {action}"
+        
+        return result
+    
+    def post(self, shared, prep_res, exec_res):
+        """Save action result"""
+        # Save the current action result
+        shared["current_action_result"] = exec_res
+        print(f"✅ Action completed, result obtained")
+        
+        # Continue to observation node
+        return "observe"
+    
+    # Simulated tool functions
+    def search_web(self, query):
+        # This should be actual search logic
+        return f"Search results: Information about '{query}'..."
+    
+    def calculate(self, expression):
+        # This should be actual calculation logic
+        try:
+            return f"Calculation result: {eval(expression)}"
+        except:
+            return f"Unable to calculate expression: {expression}"
+
+class ObserveNode(Node):
+    def prep(self, shared):
+        """Prepare observation data"""
+        action = shared["current_action"]
+        action_input = shared["current_action_input"]
+        action_result = shared["current_action_result"]
+        return action, action_input, action_result
+    
+    def exec(self, inputs):
+        """Analyze action results, generate observation"""
+        action, action_input, action_result = inputs
+        
+        # Build prompt
+        prompt = f"""
+        You are an observer, needing to analyze action results and provide objective observations.
+        
+        Action: {action}
+        Action input: {action_input}
+        Action result: {action_result}
+        
+        Please provide a concise observation of this result. Don't make decisions, just describe what you see.
+        """
+        
+        # Call LLM to get observation result
+        observation = call_llm(prompt)
+        
+        print(f"👁️ Observation: {observation[:50]}...")
+        return observation
+    
+    def post(self, shared, prep_res, exec_res):
+        """Save observation result and decide next flow step"""
+        # Save observation result
+        if "observations" not in shared:
+            shared["observations"] = []
+        shared["observations"].append(exec_res)
+        
+        # Continue thinking
+        return "think"
+    
+
+    
+class EndNode(Node):
+    def prep(self, shared):
+        """Prepare end node"""
+        
+        return {}
+    def exec(self, prep_res):
+        """Execute end operation"""
+        print("Flow ended, thank you for using!")
+        return None
+    def post(self, shared, prep_res, exec_res):
+        """End flow"""
+        return None
diff --git a/cookbook/pocketflow-tao/utils.py b/cookbook/pocketflow-tao/utils.py
@@ -0,0 +1,19 @@
+# utils.py
+
+from openai import OpenAI
+import os
+
+def call_llm(prompt):    
+    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "Your Key Here"),base_url=os.environ.get("OPENAI_API_BASE", "Your API Base Here"))
+    r = client.chat.completions.create(
+        model=os.environ.get("OPENAI_MODEL", "openai/gpt-4.1-nano"),
+        messages=[{"role": "user", "content": prompt}]
+    )
+    return r.choices[0].message.content
+
+if __name__ == "__main__":
+    print("## Testing call_llm")
+    prompt = "In a few words, what is the meaning of life?"
+    print(f"## Prompt: {prompt}")
+    response = call_llm(prompt)
+    print(f"## Response: {response}")
