|
|
|
""" |
|
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() |