| |
| """ |
| Update deprecated langchain imports to langchain_community in project files |
| """ |
| import os |
| import re |
|
|
| def update_imports(file_path): |
| """Update imports in a single file""" |
| print(f"Processing {file_path}") |
| |
| with open(file_path, 'r', encoding='utf-8') as file: |
| content = file.read() |
| |
| |
| replacements = [ |
| (r'from langchain\.vectorstores import (.*)', r'from langchain_community.vectorstores import \1'), |
| (r'from langchain\.llms import (.*)', r'from langchain_community.llms import \1'), |
| (r'from langchain\.document_loaders import (.*)', r'from langchain_community.document_loaders import \1'), |
| (r'from langchain\.embeddings import (.*)', r'from langchain_community.embeddings import \1'), |
| (r'import langchain\.vectorstores', r'import langchain_community.vectorstores'), |
| (r'import langchain\.llms', r'import langchain_community.llms'), |
| (r'import langchain\.document_loaders', r'import langchain_community.document_loaders'), |
| (r'import langchain\.embeddings', r'import langchain_community.embeddings'), |
| ] |
| |
| |
| for pattern, replacement in replacements: |
| content = re.sub(pattern, replacement, content) |
| |
| |
| with open(file_path, 'w', encoding='utf-8') as file: |
| file.write(content) |
| |
| return True |
|
|
| def main(): |
| """Main function to update all files""" |
| |
| files_to_update = [ |
| 'app/core/memory.py', |
| 'app/core/llm.py', |
| 'app/core/ingestion.py', |
| 'app/core/agent.py' |
| ] |
| |
| updated_count = 0 |
| for file_path in files_to_update: |
| if os.path.exists(file_path): |
| success = update_imports(file_path) |
| if success: |
| updated_count += 1 |
| else: |
| print(f"File not found: {file_path}") |
| |
| print(f"\nCompleted! Updated {updated_count}/{len(files_to_update)} files.") |
|
|
| if __name__ == "__main__": |
| main() |