How to Use LangChain with OpenAI API (ChatGPT + Tools)

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    model="gpt-4o-mini",           # Model name
    temperature=0.7,                # 0=deterministic, 1=creative
    max_tokens=2048,                # Max output tokens (not input)
    max_retries=3,                  # Auto-retry on rate limits / errors
    request_timeout=60,             # Seconds before timeout
    streaming=False,                # Enable for real-time output
    
    # Advanced options
    model_kwargs={
        "top_p": 0.95,              # Nucleus sampling parameter
        "frequency_penalty": 0.1,   # Reduce repetition (0 to 2)
        "presence_penalty": 0.0,    # Topic diversity (0 to 2)
        "seed": 42,                 # Deterministic outputs (when temp=0)
    },
    
    # For Azure OpenAI deployments
    # openai_api_base="https://your-resource.openai.azure.com/",
    # openai_api_version="2024-02-01",
    # azure_deployment="gpt-4o-mini",
)

# Quick test
from langchain_core.messages import HumanMessage
response = llm.invoke([HumanMessage(content="Say 'hello' in five languages.")])
print(response.content)

from langchain_core.callbacks import UsageMetadataCallbackHandler
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# Track token usage
callback = UsageMetadataCallbackHandler()
llm = ChatOpenAI(model="gpt-4o-mini")

response = llm.invoke(
    [HumanMessage(content="Explain gradient descent in 100 words.")],
    config={"callbacks": [callback]}
)

print(response.content)
# Access usage data from the response directly
if hasattr(response, 'usage_metadata') and response.usage_metadata:
    print(f"Input tokens: {response.usage_metadata['input_tokens']}")
    print(f"Output tokens: {response.usage_metadata['output_tokens']}")
    print(f"Total tokens: {response.usage_metadata['total_tokens']}")

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import json

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Define functions using OpenAI's JSON schema format
functions = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a city",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "The city name"
                },
                "units": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature units"
                }
            },
            "required": ["city"]
        }
    }
]

# Bind functions to the model
llm_with_functions = llm.bind(functions=functions)

response = llm_with_functions.invoke(
    [HumanMessage(content="What's the weather in Tokyo?")]
)

# Check if model wants to call a function
if response.additional_kwargs.get("function_call"):
    func_call = response.additional_kwargs["function_call"]
    print(f"Function name: {func_call['name']}")
    print(f"Arguments: {func_call['arguments']}")
    args = json.loads(func_call['arguments'])
    print(f"City: {args['city']}")

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from typing import Optional
from pydantic import BaseModel, Field

# Define tools with the @tool decorator
@tool
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a given city.
    
    Args:
        city: The city name (e.g., 'Tokyo', 'New York')
        units: Temperature units, either 'celsius' or 'fahrenheit'
    """
    # Real implementation would call a weather API
    return f"Weather in {city}: 22°C (clear sky)"

@tool
def search_products(
    query: str, 
    category: Optional[str] = None,
    max_price: Optional[float] = None
) -> str:
    """Search for products in the store catalog.
    
    Args:
        query: Search query string
        category: Optional product category filter
        max_price: Optional maximum price filter
    """
    return f"Found 5 products matching '{query}' in category '{category}'"

@tool
def calculate_shipping(
    weight_kg: float,
    destination_country: str,
    express: bool = False
) -> str:
    """Calculate shipping cost for an order.
    
    Args:
        weight_kg: Package weight in kilograms
        destination_country: Two-letter country code (e.g., 'US', 'DE', 'JP')
        express: Whether to use express shipping
    """
    base_cost = weight_kg * 2.5
    if express:
        base_cost *= 2.5
    return f"Shipping to {destination_country}: ${base_cost:.2f}"

# Bind multiple tools at once
tools = [get_weather, search_products, calculate_shipping]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# The model now knows about all three tools
response = llm_with_tools.invoke(
    [HumanMessage(content="I want to buy a laptop under $800 and ship it to Germany. How much will shipping cost for a 2kg package?")]
)

print(f"Tool calls: {response.tool_calls}")

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from typing import Any
import json

# Define tools
@tool
def web_search(query: str) -> str:
    """Search the web for current information."""
    # Mock — real version uses DuckDuckGo, Serper, or Bing API
    return f"Search results for '{query}': [Result 1: Python 3.13 released...] [Result 2: ...]"

@tool
def run_python_code(code: str) -> str:
    """Execute Python code and return the output. Safe for math and string operations."""
    try:
        # In production, use a sandboxed environment
        result = eval(code)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool  
def save_note(content: str, title: str) -> str:
    """Save a note for later reference.
    
    Args:
        content: The note content
        title: A short descriptive title
    """
    # In production, write to a database or file
    print(f"[NOTE SAVED] {title}: {content[:50]}...")
    return f"Note '{title}' saved successfully."

tools = [web_search, run_python_code, save_note]
tool_map = {t.name: t for t in tools}

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

def run_tool_loop(user_message: str) -> str:
    """Run the complete tool-calling loop."""
    messages = [HumanMessage(content=user_message)]
    
    while True:
        response = llm_with_tools.invoke(messages)
        messages.append(response)
        
        # If no tool calls, we have the final answer
        if not response.tool_calls:
            return response.content
        
        # Execute all requested tools
        for tool_call in response.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            
            if tool_name in tool_map:
                result = tool_map[tool_name].invoke(tool_args)
            else:
                result = f"Error: Unknown tool '{tool_name}'"
            
            # Add tool result to messages
            messages.append(
                ToolMessage(
                    content=str(result),
                    tool_call_id=tool_call["id"]
                )
            )
            
            print(f"[Tool: {tool_name}] Input: {tool_args}")
            print(f"[Tool: {tool_name}] Output: {result[:100]}")

# Test the loop
answer = run_tool_loop(
    "Search for the latest Python version, then calculate 2 to the power of the version number's major version, and save the result as a note."
)
print(f"\nFinal Answer: {answer}")

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

@tool
def get_stock_price(ticker: str) -> str:
    """Get the current stock price for a ticker symbol."""
    prices = {"AAPL": "$182.50", "GOOGL": "$178.20", "MSFT": "$415.00"}
    return prices.get(ticker.upper(), f"Unknown ticker: {ticker}")

@tool
def get_company_info(ticker: str) -> str:
    """Get basic company information for a ticker."""
    info = {
        "AAPL": "Apple Inc. — Consumer electronics and software",
        "GOOGL": "Alphabet Inc. — Technology conglomerate",
        "MSFT": "Microsoft Corp. — Software and cloud services"
    }
    return info.get(ticker.upper(), f"No info for: {ticker}")

tools = [get_stock_price, get_company_info]
tool_map = {t.name: t for t in tools}

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# This query should trigger parallel calls for both tools
messages = [HumanMessage(content="Get the stock price AND company info for Apple, all at once.")]
response = llm_with_tools.invoke(messages)

print(f"Number of tool calls: {len(response.tool_calls)}")
# With gpt-4o or gpt-4o-mini, this should be 2 simultaneous calls

# Execute all tool calls (they can be parallelized in your code too)
import asyncio
from langchain_core.tools import BaseTool

async def execute_tool_calls_parallel(tool_calls: list) -> list[ToolMessage]:
    async def run_one(tc):
        tool = tool_map[tc["name"]]
        result = await asyncio.to_thread(tool.invoke, tc["args"])
        return ToolMessage(content=str(result), tool_call_id=tc["id"])
    
    return await asyncio.gather(*[run_one(tc) for tc in tool_calls])

tool_messages = asyncio.run(execute_tool_calls_parallel(response.tool_calls))
messages.append(response)
messages.extend(tool_messages)

final = llm_with_tools.invoke(messages)
print(f"\nFinal response: {final.content}")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Route between models based on task type
fast_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
smart_llm = ChatOpenAI(model="gpt-4o", temperature=0)

classifier_prompt = ChatPromptTemplate.from_template(
    """Classify this task as 'simple' or 'complex'.
    Simple: factual Q&A, basic summaries, formatting
    Complex: multi-step reasoning, code generation, analysis
    
    Task: {task}
    
    Respond with ONE word: 'simple' or 'complex'"""
)

classifier = classifier_prompt | fast_llm | StrOutputParser()

def smart_route(task: str) -> str:
    complexity = classifier.invoke({"task": task}).strip().lower()
    
    if complexity == "complex":
        llm = smart_llm
        print("[Using gpt-4o for complex task]")
    else:
        llm = fast_llm
        print("[Using gpt-4o-mini for simple task]")
    
    task_prompt = ChatPromptTemplate.from_template("Complete this task:\n{task}")
    chain = task_prompt | llm | StrOutputParser()
    return chain.invoke({"task": task})

print(smart_route("What is 2 + 2?"))
print(smart_route("Analyze the time complexity of merge sort and compare it to quicksort with mathematical proof."))

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List

class ProductReview(BaseModel):
    sentiment: str = Field(description="positive, negative, or neutral")
    score: int = Field(description="Score from 1-10", ge=1, le=10)
    key_points: List[str] = Field(description="3-5 key points from the review")
    topics_mentioned: List[str] = Field(description="Product aspects mentioned")
    would_recommend: bool = Field(description="Whether reviewer recommends the product")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(ProductReview)

review_text = """
This laptop is absolutely amazing. The battery life is incredible - I get 
12 hours easily. The keyboard feels great and the screen is stunning. 
Only minor complaint is it runs a bit warm under heavy load. 
Definitely recommending this to my friends.
"""

analysis = structured_llm.invoke(f"Analyze this review:\n\n{review_text}")
print(f"Sentiment: {analysis.sentiment}")
print(f"Score: {analysis.score}/10")
print(f"Key Points: {analysis.key_points}")
print(f"Would Recommend: {analysis.would_recommend}")

from langchain_openai import ChatOpenAI
import time

# LangChain handles basic retries automatically
llm = ChatOpenAI(
    model="gpt-4o-mini",
    max_retries=5,
    request_timeout=60,
)

# For batch processing with rate limit awareness
async def process_batch_with_rate_limit(prompts: list, delay: float = 0.1):
    """Process a batch of prompts respecting rate limits."""
    import asyncio
    from langchain_core.messages import HumanMessage
    
    results = []
    for prompt in prompts:
        response = await llm.ainvoke([HumanMessage(content=prompt)])
        results.append(response.content)
        await asyncio.sleep(delay)  # Basic rate limiting
    return results

from langchain_community.cache import SQLiteCache
from langchain_core.globals import set_llm_cache

# Persistent cache — identical prompts won't hit the API
set_llm_cache(SQLiteCache(database_path=".openai_cache.db"))

from openai import BadRequestError
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")

def safe_invoke(message: str) -> str:
    try:
        response = llm.invoke([HumanMessage(content=message)])
        return response.content
    except BadRequestError as e:
        if "content_policy_violation" in str(e):
            return "I can't help with that request."
        raise  # Re-raise unexpected errors
    except Exception as e:
        return f"An error occurred: {str(e)}"