Building a Foreign Correspondent With MongoDB, Anthropic's Claude, Python

1
python -m venv venv

1
anthropic==0.38.0
2
ipykernel==6.29.5
3
ipython==8.28.0
4
pandas==2.2.3
5
pymongo==4.10.1
6
python-dotenv==1.0.1
7
scikit-learn==1.5.2
8
tqdm==4.66.5
9
transformers==4.46.0

1
pip install -r requirements.txt

1
import json
2
import logging
3
import os
4
import time
5
from datetime import datetime
6
from typing import Dict, List, Optional, Tuple
7
8
# Third-party
9
import pandas as pd
10
from anthropic import Anthropic
11
from dotenv import load_dotenv
12
from pymongo import MongoClient
13
from pymongo.errors import BulkWriteError, ConnectionFailure
14
from sentence_transformers import SentenceTransformer
15
from tqdm.autonotebook import tqdm, trange

1
articles = pd.read_csv('fa_articles.csv', encoding='utf-8')
2
articles.head()

1
# Initialize Embedić model for Serbian language embeddings
2
# !pip install -U sentence-transformers
3
from sentence_transformers import SentenceTransformer
4
from tqdm.notebook import tqdm
5
from datetime import datetime
6
from typing import List, Dict
7
import json
8
9
model = SentenceTransformer('djovak/embedic-large')

1
def generate_embeddings(df: pd.DataFrame, batch_size: int = 32) -> pd.DataFrame:
2
    """
3
    Generate embeddings for concatenated title and content using Embedić.
4
    
5
    Args:
6
        df: DataFrame with 'title' and 'content' columns
7
        batch_size: Number of texts to process at once
8
        
9
    Returns:
10
        DataFrame with added 'embedding' column
11
    """
12
    # Create a copy of the dataframe
13
    df_emb = df.copy()
14
    
15
    # Concatenate title and content
16
    print("Concatenating title and content...")
17
    df_emb['text_for_embedding'] = df_emb['title'] + " " + df_emb['content']
18
    
19
    # Generate embeddings in batches
20
    print("Generating embeddings...")
21
    embeddings = []
22
    
23
    # Convert texts to list for batch processing
24
    texts = df_emb['text_for_embedding'].tolist()
25
    
26
    # Process in batches with progress bar
27
    for i in tqdm(range(0, len(texts), batch_size)):
28
        batch_text = texts[i:i + batch_size]
29
        batch_embeddings = model.encode(batch_text)
30
        embeddings.extend(batch_embeddings.tolist())
31
    
32
    # Add embeddings to dataframe
33
    df_emb['embedding'] = embeddings
34
    
35
    # Remove temporary concatenated text column
36
    df_emb = df_emb.drop('text_for_embedding', axis=1)
37
    
38
    print(f"Generated embeddings for {len(df_emb)} articles")
39
    print(f"Embedding dimension: {len(embeddings[0])}")
40
    
41
    return df_emb

1
print("\nDataFrame shape:", df_with_embeddings.shape)
2
print("\nColumns:", df_with_embeddings.columns.tolist())
3
4
print("\nSample embedding (first 5 dimensions):")
5
print(df_with_embeddings['embedding'].iloc[0][:5])
6
7
embedding_lengths = df_with_embeddings['embedding'].apply(len)
8
print("\nAll embeddings have same dimension:", embedding_lengths.nunique() == 1)
9
print("Embedding dimension:", embedding_lengths.iloc[0])

1
def prepare_for_mongodb(df: pd.DataFrame) -> List[Dict]:
2
    documents = []
3
    
4
    for _, row in tqdm(df.iterrows(), total=len(df)):
5
        document = {           
6
            'title': row['title'],
7
            'content': row['content'],
8
            'date': row['artDate'],  # Assuming you have a date column
9
            'embedding': row['embedding'],
10
            'metadata': {
11
                'embedding_model': 'djovak/embedic-large',
12
                'created_at': datetime.now(),
13
                'last_updated': datetime.now()
14
            }
15
        }
16
        documents.append(document)
17
    
18
    return documents

1
import os
2
from datetime import datetime
3
from dotenv import load_dotenv
4
from pymongo import MongoClient
5
from typing import Dict, List, Optional, Tuple
6
import logging
7
8
load_dotenv()
9
MONGODB_URI = os.getenv("MONGODB_URI")
10
DB_NAME = os.getenv("DB_NAME")
11
COLLECTION_NAME = os.getenv("COLLECTION_NAME")
12
13
logging.basicConfig(level=logging.INFO)
14
logger = logging.getLogger(__name__)

1
MONGODB_URI = "youURI"
2
COLLECTION_NAME="articles"
3
DB_NAME="rag_articles"
4
ANTHROPIC_API_KEY=sk-xxx…

1
def get_mongodb_connection():
2
    try:
3
        client = MongoClient(MONGODB_URI)
4
        client.admin.command('ping')
5
        logger.info("Successfully connected to MongoDB")
6
        
7
        db = client[DB_NAME]
8
        collection = db[COLLECTION_NAME]
9
        return client, db, collection
10
        
11
    except Exception as e:
12
        logger.error(f"Failed to connect to MongoDB. Error: {str(e)}")
13
        raise ConnectionFailure(f"MongoDB connection failed: {str(e)}")
14
15
client, db, collection = get_mongodb_connection()

1
def insert_documents_to_mongodb(documents: List[Dict], batch_size: int = 1000)->Tuple[int, int]:
2
    successful = 0
3
    failed = 0
4
5
    for i in tqdm(range(0, len(documents), batch_size)):
6
        batch = documents[i:i + batch_size]
7
        try:
8
            # Insert batch with ordered=False for better performance
9
            result = collection.insert_many(batch, ordered=False)
10
            successful += len(result.inserted_ids)
11
        except BulkWriteError as e:
12
            # Handle partial failures in batch
13
            successful += e.details['nInserted']
14
            failed += len(batch) - e.details['nInserted']
15
            print(f"Batch {i//batch_size + 1} had {e.details['nInserted']} successful and "
16
                  f"{len(batch) - e.details['nInserted']} failed insertions")
17
    
18
    return successful, failed

1
collection.delete_many({})
2
# Insert documents
3
successful, failed = insert_documents_to_mongodb(mongodb_documents)
4
print(f"Successfully inserted: {successful} documents")
5
print(f"Failed insertions: {failed} documents")

1
def create_vector_index(
2
    connection_string: str,
3
    database_name: str,
4
    collection_name: str,
5
    index_name: str = "vector_index"
6
) -> None:
7
    """
8
    Create a vector search index in MongoDB using PyMongo.
9
    
10
    Args:
11
        connection_string: MongoDB connection string
12
        database_name: Name of the database
13
        collection_name: Name of the collection
14
        index_name: Name of the vector index
15
    """
16
    # Connect to MongoDB
17
    client = client
18
    
19
    # Get database and collection
20
    db = db
21
    collection = collection
22
    
23
    # Define the index configuration
24
    index_config = {
25
        "name": index_name,
26
        "definition": {
27
            "fields": [
28
                {
29
                    "numDimensions": 1024,
30
                    "path": "embedding",
31
                    "similarity": "cosine",
32
                    "type": "vector"
33
                }
34
            ]
35
        }
36
    }
37
    
38
    try:
39
        # Create the vector search index
40
        collection.create_search_index(index_config)
41
        print(f"Successfully created vector index '{index_name}'")
42
        
43
    except Exception as e:
44
        print(f"Error creating vector index: {str(e)}")        
45
    finally:
46
        client.close()

1
def semantic_search(collection, query: str, k: int = 5) -> List[Dict]:
2
    """
3
    Perform semantic search using Embedić vector similarity.
4
    
5
    Args:
6
        collection: MongoDB collection
7
        query: Search query text - in Serbian!
8
        k: Number of results to return
9
        
10
    Returns:
11
        List of matching documents
12
    """
13
    try:
14
        # Generate embedding for query using Embedić
15
        query_embedding = model.encode(query).tolist()
16
        
17
        pipeline = [
18
            {
19
                "$vectorSearch": {
20
                    "index": "vector_index",
21
                    "queryVector": query_embedding,
22
                    "path": "embedding",
23
                    "numCandidates": 100,
24
                    "limit": k
25
                }
26
            },
27
            {
28
                "$project": {
29
                    "title": 1,
30
                    "content": 1,
31
                    "date": 1,             
32
                    "score": { "$meta": "vectorSearchScore" }
33
                }
34
            }
35
        ]
36
        
37
        results = list(collection.aggregate(pipeline))
38
        return results
39
        
40
    except Exception as e:
41
        logger.error(f"Search error: {str(e)}")
42
        raise

1
def display_search_results(query: str, results: List[Dict]):
2
    """
3
    Display search results in a readable format
4
    """
5
    print(f"\nQuery: {query}")
6
    print("=" * 80)
7
    
8
    for i, doc in enumerate(results, 1):
9
        print(f"\n{i}. {doc['title']}")
10
        print(f"Score: {doc['score']:.3f}")
11
        print("-" * 80)
12
13
# Let's test the search
14
queries = [
15
    "Novi Sad","Litijum", "Ekonomija"
16
]
17
18
for query in queries:
19
    results = semantic_search(collection, query, k=6)
20
    display_search_results(query, results)
21
    print("\n" + "=" * 100 + "\n")

1
#!pip install anthropic
2
from anthropic import Anthropic
3
import os
4
from typing import List, Dict, Optional
5
import time
6
7
load_dotenv()
8
9
# Initialize Claude client
10
anthropic = Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))

1
def generate_news_report(
2
    query: str,
3
    collection,
4
    model: SentenceTransformer,
5
    anthropic_client: Anthropic,
6
    language: str = "English",
7
    top_k: int = 10,
8
    max_retries: int = 3,
9
    retry_delay: int = 2
10
) -> str:
11
    # Translate query to Serbian using Claude
12
    translation_prompt = f"Translate the following {language} text to Serbian latin. Provide only the translation, nothing else: '{query}'"    
13
    try:
14
        translation_response = anthropic_client.messages.create(
15
            model="claude-3-opus-20240229",
16
            max_tokens=100,
17
            temperature=0,
18
            messages=[{
19
                "role": "user",
20
                "content": translation_prompt
21
            }]
22
        )
23
        
24
        serbian_query = translation_response.content[0].text.strip()
25
    except Exception as e:
26
        raise Exception(f"Translation failed: {str(e)}")
27
    
28
    # Generate embedding for Serbian query
29
    query_embedding = model.encode(serbian_query).tolist()
30
    
31
    # Perform vector search
32
    pipeline = [
33
        {
34
            "$vectorSearch": {
35
                "index": "vector_index",
36
                "queryVector": query_embedding,
37
                "path": "embedding",
38
                "numCandidates": 100,
39
                "limit": top_k
40
            }
41
        },
42
        {
43
            "$project": {
44
                "title": 1,
45
                "content": 1,
46
                "date": 1,
47
                "score": { "$meta": "vectorSearchScore" }
48
            }
49
        }
50
    ]
51
    
52
    results = list(collection.aggregate(pipeline))
53
    
54
    # Prepare content for Claude
55
    articles_text = "\n\n".join([
56
        f"Article {i+1}:\nDate: {doc['date']}\nTitle: {doc['title']}\nContent: {doc['content'][:500]}..."
57
        for i, doc in enumerate(results)
58
    ])
59
    
60
    # Create prompt for report generation
61
    prompt = f"""You are an expert journalist and news analyst. Based on the following {top_k} most relevant Serbian news articles about "{query}" (translated to Serbian as "{serbian_query}"), 
62
create a concise, well-structured news report in {language}. The report should:
63
- Be around 250-300 words
64
- Start with a clear headline
65
- Include key facts, dates, and relevant context
66
- Maintain journalistic neutrality
67
- Focus on the most newsworthy aspects
68
- Include a brief conclusion or outlook
69
Here are the articles:
70
{articles_text}
71
Please write the report in a professional journalistic style."""
72
73
    # Get response from Claude with retries
74
    for attempt in range(max_retries):
75
        try:
76
            response = anthropic_client.messages.create(
77
                model="claude-3-opus-20240229",
78
                max_tokens=1000,
79
                temperature=0.3,
80
                messages=[{
81
                    "role": "user",
82
                    "content": prompt
83
                }]
84
            )
85
            
86
            report = response.content[0].text.strip()
87
            return report
88
            
89
        except Exception as e:
90
            if attempt == max_retries - 1:
91
                raise Exception(f"Failed to generate report: {str(e)}")
92
            time.sleep(retry_delay)
93
            continue

1
try:
2
    query = "Investments in mining"
3
4
    report = generate_news_report(
5
        query=query,
6
        collection=collection,
7
        model=model,  # our existing sentence transformer model
8
        anthropic_client=anthropic,
9
        top_k=10,
10
    )
11
12
    print("\nGenerated News Report:")
13
    print("=" * 80)
14
    print(report)
15
16
except Exception as e:
17
    print(f"Error generating report: {str(e)}")