Create reasoning_engines.py
Browse files- reasoning_engines.py +50 -0
reasoning_engines.py
ADDED
|
@@ -0,0 +1,50 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
# [Use the CoT/ToT implementations from ethical-rag-starter.py]
|
| 2 |
+
# Minimal version for quick deployment:
|
| 3 |
+
|
| 4 |
+
import re
|
| 5 |
+
from typing import Tuple, List
|
| 6 |
+
|
| 7 |
+
class ChainOfThought:
|
| 8 |
+
def __init__(self, llm):
|
| 9 |
+
self.llm = llm
|
| 10 |
+
|
| 11 |
+
def basic_cot(self, query: str) -> Tuple[str, List[str]]:
|
| 12 |
+
prompt = f"""Think step-by-step about: {query}
|
| 13 |
+
|
| 14 |
+
Step 1: Understand the question
|
| 15 |
+
Step 2: Break into parts
|
| 16 |
+
Step 3: Solve each part
|
| 17 |
+
Step 4: Combine
|
| 18 |
+
|
| 19 |
+
Provide your answer:"""
|
| 20 |
+
|
| 21 |
+
response = self.llm.generate(prompt)
|
| 22 |
+
steps = [line.strip() for line in response.split('\n')
|
| 23 |
+
if line.strip() and any(f"Step {i}" in line for i in range(1,5))]
|
| 24 |
+
|
| 25 |
+
return response, steps
|
| 26 |
+
|
| 27 |
+
def self_consistency_cot(self, query: str, num_paths: int = 2) -> Tuple[str, float]:
|
| 28 |
+
responses = []
|
| 29 |
+
for i in range(num_paths):
|
| 30 |
+
response = self.llm.generate(f"Problem: {query}\n\nSolution {i+1}:")
|
| 31 |
+
responses.append(response)
|
| 32 |
+
|
| 33 |
+
return responses, 0.7
|
| 34 |
+
|
| 35 |
+
class TreeOfThoughts:
|
| 36 |
+
def __init__(self, llm, max_depth=2, branching_factor=2):
|
| 37 |
+
self.llm = llm
|
| 38 |
+
self.max_depth = max_depth
|
| 39 |
+
self.branching_factor = branching_factor
|
| 40 |
+
|
| 41 |
+
def solve_bfs(self, problem: str) -> Tuple[str, List[str], List]:
|
| 42 |
+
# Simplified BFS
|
| 43 |
+
branches = self.llm.generate(
|
| 44 |
+
f"List {self.branching_factor} approaches to: {problem}"
|
| 45 |
+
)
|
| 46 |
+
return branches, [], []
|
| 47 |
+
|
| 48 |
+
def solve_dfs(self, problem: str) -> Tuple[str, List[str], List]:
|
| 49 |
+
# Simplified DFS
|
| 50 |
+
return self.llm.generate(f"Solve: {problem}"), [], []
|