saberbud/PocketFlow-Tutorial-Codebase-Knowledge
1
0
Fork 0
mirror of https://github.com/The-Pocket/PocketFlow-Tutorial-Codebase-Knowledge.git synced 2026-08-30 09:00:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/main' into crawling-progress

Browse Source
This commit is contained in:
Hari Haran
2025-05-05 10:39:02 +00:00
parent cd947bbdc6 cd9e2b9804
commit 0b2078b12a
9 changed files with 537 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/call_llm.py
+110 -28
View File
@@ -3,86 +3,93 @@ import os
import logging
import json
from datetime import datetime
import requests
# Configure logging
log_directory = os.getenv("LOG_DIR", "logs")
os.makedirs(log_directory, exist_ok=True)
log_file = os.path.join(log_directory, f"llm_calls_{datetime.now().strftime('%Y%m%d')}.log")
log_file = os.path.join(
log_directory, f"llm_calls_{datetime.now().strftime('%Y%m%d')}.log"
)
# Set up logger
logger = logging.getLogger("llm_logger")
logger.setLevel(logging.INFO)
logger.propagate = False # Prevent propagation to root logger
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
file_handler.setFormatter(
logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
)
logger.addHandler(file_handler)
# Simple cache configuration
cache_file = "llm_cache.json"
# By default, we Google Gemini 2.5 pro, as it shows great performance for code understanding
def call_llm(prompt: str, use_cache: bool = True) -> str:
# Log the prompt
logger.info(f"PROMPT: {prompt}")
# Check cache if enabled
if use_cache:
# Load cache from disk
cache = {}
if os.path.exists(cache_file):
try:
with open(cache_file, 'r') as f:
with open(cache_file, "r") as f:
cache = json.load(f)
except:
logger.warning(f"Failed to load cache, starting with empty cache")
# Return from cache if exists
if prompt in cache:
logger.info(f"RESPONSE: {cache[prompt]}")
return cache[prompt]
# Call the LLM if not in cache or cache disabled
client = genai.Client(
vertexai=True,
# TODO: change to your own project id and location
project=os.getenv("GEMINI_PROJECT_ID", "your-project-id"),
location=os.getenv("GEMINI_LOCATION", "us-central1")
)
# You can comment the previous line and use the AI Studio key instead:
# # Call the LLM if not in cache or cache disabled
# client = genai.Client(
# api_key=os.getenv("GEMINI_API_KEY", "your-api_key"),
# vertexai=True,
# # TODO: change to your own project id and location
# project=os.getenv("GEMINI_PROJECT_ID", "your-project-id"),
# location=os.getenv("GEMINI_LOCATION", "us-central1")
# )
model = os.getenv("GEMINI_MODEL", "gemini-2.5-pro-exp-03-25")
response = client.models.generate_content(
model=model,
contents=[prompt]
# You can comment the previous line and use the AI Studio key instead:
client = genai.Client(
api_key=os.getenv("GEMINI_API_KEY", ""),
)
response_text = response.text
model = os.getenv("GEMINI_MODEL", "gemini-2.5-pro-exp-03-25")
# model = os.getenv("GEMINI_MODEL", "gemini-2.5-flash-preview-04-17")
response = client.models.generate_content(model=model, contents=[prompt])
response_text = response.text
# Log the response
logger.info(f"RESPONSE: {response_text}")
# Update cache if enabled
if use_cache:
# Load cache again to avoid overwrites
cache = {}
if os.path.exists(cache_file):
try:
with open(cache_file, 'r') as f:
with open(cache_file, "r") as f:
cache = json.load(f)
except:
pass
# Add to cache and save
cache[prompt] = response_text
try:
with open(cache_file, 'w') as f:
with open(cache_file, "w") as f:
json.dump(cache, f)
except Exception as e:
logger.error(f"Failed to save cache: {e}")
return response_text
# # Use Anthropic Claude 3.7 Sonnet Extended Thinking
# def call_llm(prompt, use_cache: bool = True):
# from anthropic import Anthropic
@@ -101,7 +108,7 @@ def call_llm(prompt: str, use_cache: bool = True) -> str:
# return response.content[1].text
# # Use OpenAI o1
# def call_llm(prompt, use_cache: bool = True):
# def call_llm(prompt, use_cache: bool = True):
# from openai import OpenAI
# client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "your-api-key"))
# r = client.chat.completions.create(
@@ -115,11 +122,86 @@ def call_llm(prompt: str, use_cache: bool = True) -> str:
# )
# return r.choices[0].message.content
# Use OpenRouter API
# def call_llm(prompt: str, use_cache: bool = True) -> str:
# # Log the prompt
# logger.info(f"PROMPT: {prompt}")
# # Check cache if enabled
# if use_cache:
# # Load cache from disk
# cache = {}
# if os.path.exists(cache_file):
# try:
# with open(cache_file, "r") as f:
# cache = json.load(f)
# except:
# logger.warning(f"Failed to load cache, starting with empty cache")
# # Return from cache if exists
# if prompt in cache:
# logger.info(f"RESPONSE: {cache[prompt]}")
# return cache[prompt]
# # OpenRouter API configuration
# api_key = os.getenv("OPENROUTER_API_KEY", "")
# model = os.getenv("OPENROUTER_MODEL", "google/gemini-2.0-flash-exp:free")
# headers = {
# "Authorization": f"Bearer {api_key}",
# }
# data = {
# "model": model,
# "messages": [{"role": "user", "content": prompt}]
# }
# response = requests.post(
# "https://openrouter.ai/api/v1/chat/completions",
# headers=headers,
# json=data
# )
# if response.status_code != 200:
# error_msg = f"OpenRouter API call failed with status {response.status_code}: {response.text}"
# logger.error(error_msg)
# raise Exception(error_msg)
# try:
# response_text = response.json()["choices"][0]["message"]["content"]
# except Exception as e:
# error_msg = f"Failed to parse OpenRouter response: {e}; Response: {response.text}"
# logger.error(error_msg)
# raise Exception(error_msg)
# # Log the response
# logger.info(f"RESPONSE: {response_text}")
# # Update cache if enabled
# if use_cache:
# # Load cache again to avoid overwrites
# cache = {}
# if os.path.exists(cache_file):
# try:
# with open(cache_file, "r") as f:
# cache = json.load(f)
# except:
# pass
# # Add to cache and save
# cache[prompt] = response_text
# try:
# with open(cache_file, "w") as f:
# json.dump(cache, f)
# except Exception as e:
# logger.error(f"Failed to save cache: {e}")
# return response_text
if __name__ == "__main__":
test_prompt = "Hello, how are you?"
# First call - should hit the API
print("Making call...")
response1 = call_llm(test_prompt, use_cache=False)
print(f"Response: {response1}")
utils/crawl_local_files.py
+70 -25
View File
@@ -1,10 +1,17 @@
import os
import fnmatch
import pathspec
def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, max_file_size=None, use_relative_paths=True, progress_callback=None):
def crawl_local_files(
directory,
include_patterns=None,
exclude_patterns=None,
max_file_size=None,
use_relative_paths=True,
progress_callback=None,
):
"""
Crawl files in a local directory with similar interface as crawl_github_files.
Args:
directory (str): Path to local directory
include_patterns (set): File patterns to include (e.g. {"*.py", "*.js"})
@@ -12,18 +19,48 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
max_file_size (int): Maximum file size in bytes
use_relative_paths (bool): Whether to use paths relative to directory
progress_callback (callable): Function to report progress, takes (processed, total) as arguments
Returns:
dict: {"files": {filepath: content}}
"""
if not os.path.isdir(directory):
raise ValueError(f"Directory does not exist: {directory}")
files_dict = {}
all_files = []
# Collect all files first to calculate total
for root, _, files in os.walk(directory):
files_dict = {}
# --- Load .gitignore ---
gitignore_path = os.path.join(directory, ".gitignore")
gitignore_spec = None
if os.path.exists(gitignore_path):
try:
with open(gitignore_path, "r", encoding="utf-8") as f:
gitignore_patterns = f.readlines()
gitignore_spec = pathspec.PathSpec.from_lines("gitwildmatch", gitignore_patterns)
print(f"Loaded .gitignore patterns from {gitignore_path}")
except Exception as e:
print(f"Warning: Could not read or parse .gitignore file {gitignore_path}: {e}")
all_files = []
for root, dirs, files in os.walk(directory):
# Filter directories using .gitignore and exclude_patterns early
excluded_dirs = set()
for d in dirs:
dirpath_rel = os.path.relpath(os.path.join(root, d), directory)
if gitignore_spec and gitignore_spec.match_file(dirpath_rel):
excluded_dirs.add(d)
continue
if exclude_patterns:
for pattern in exclude_patterns:
if fnmatch.fnmatch(dirpath_rel, pattern) or fnmatch.fnmatch(d, pattern):
excluded_dirs.add(d)
break
for d in dirs.copy():
if d in excluded_dirs:
dirs.remove(d)
for filename in files:
filepath = os.path.join(root, filename)
all_files.append(filepath)
@@ -32,13 +69,19 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
processed_files = 0
for filepath in all_files:
# Get path relative to directory if requested
if use_relative_paths:
relpath = os.path.relpath(filepath, directory)
else:
relpath = filepath
relpath = os.path.relpath(filepath, directory) if use_relative_paths else filepath
# --- Exclusion check ---
excluded = False
if gitignore_spec and gitignore_spec.match_file(relpath):
excluded = True
if not excluded and exclude_patterns:
for pattern in exclude_patterns:
if fnmatch.fnmatch(relpath, pattern):
excluded = True
break
# Check if file matches any include pattern
included = False
if include_patterns:
for pattern in include_patterns:
@@ -48,21 +91,12 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
else:
included = True
# Check if file matches any exclude pattern
excluded = False
if exclude_patterns:
for pattern in exclude_patterns:
if fnmatch.fnmatch(relpath, pattern):
excluded = True
break
if not included or excluded:
processed_files += 1
if progress_callback:
progress_callback(processed_files, total_files)
continue
# Check file size
if max_file_size and os.path.getsize(filepath) > max_file_size:
processed_files += 1
if progress_callback:
@@ -70,7 +104,7 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
continue
try:
with open(filepath, 'r', encoding='utf-8') as f:
with open(filepath, "r", encoding="utf-8") as f:
content = f.read()
files_dict[relpath] = content
except Exception as e:
@@ -82,9 +116,20 @@ def crawl_local_files(directory, include_patterns=None, exclude_patterns=None, m
return {"files": files_dict}
if __name__ == "__main__":
print("--- Crawling parent directory ('..') ---")
files_data = crawl_local_files("..", exclude_patterns={"*.pyc", "__pycache__/*", ".git/*", "output/*"})
files_data = crawl_local_files(
"..",
exclude_patterns={
"*.pyc",
"__pycache__/*",
".venv/*",
".git/*",
"docs/*",
"output/*",
},
)
print(f"Found {len(files_data['files'])} files:")
for path in files_data["files"]:
print(f" {path}")