Run Llama 3.3 70B in LangChain via n4n.ai

Llama 3.3 70B is the first open-weight model that genuinely competes with GPT-4o on reasoning and coding benchmarks while running on a single H100 or a pair of A100s. If you’re building with LangChain, you don’t need to manage vLLM deployments or juggle multiple provider APIs. This guide shows how to wire up Llama 3.3 70B through n4n.ai’s OpenAI-compatible endpoint, add streaming and tool calling, and harden the integration for production workloads.

Step 1: Install dependencies

Start with a clean virtual environment. You need LangChain’s OpenAI integration, the core library, and the community package for a few utilities.

python -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install "langchain-openai>=0.1.20" "langchain-core>=0.2.30" "langchain-community>=0.2.10" python-dotenv

If you’re on Python 3.12+, add pip install "pydantic>=2.7" to avoid a known validation regression.

Step 2: Configure credentials and endpoint

Create a .env file in your project root. n4n.ai exposes a single OpenAI-compatible base URL that routes to 240+ models, including Llama 3.3 70B. You only need your API key and the model identifier.

# .env
N4N_API_KEY=sk-n4n-...
N4N_BASE_URL=https://api.n4n.ai/v1
LLAMA_MODEL=meta-llama/llama-3.3-70b-instruct

The model identifier follows the provider/model-name convention. For Llama 3.3 70B Instruct, the string above is correct as of this writing. You can list available models with curl -H "Authorization: Bearer $N4N_API_KEY" $N4N_BASE_URL/models | jq '.data[].id' | grep -i llama.

Step 3: Initialize the chat model

LangChain’s ChatOpenAI class works drop-in with any OpenAI-compatible endpoint. Point it at the n4n.ai base URL and pass your key.

# llm_setup.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model=os.getenv("LLAMA_MODEL"),
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    temperature=0.2,
    max_tokens=4096,
    timeout=60,
    max_retries=2,
)

Verify it works:

# test_basic.py
from llm_setup import llm

response = llm.invoke("Write a one-sentence definition of a closure in Python.")
print(response.content)

Run python test_basic.py. You should see a coherent, technically accurate sentence. If you get a 401, check your API key. If you get a 404 on the model, verify the model ID against the models endpoint.

Step 4: Enable streaming for latency-sensitive UIs

Streaming token-by-token reduces perceived latency dramatically. LangChain supports async streaming via astream and sync via stream. Use async in FastAPI, Starlette, or any async framework.

# streaming.py
import asyncio
from llm_setup import llm

async def stream_response(prompt: str):
    async for chunk in llm.astream(prompt):
        print(chunk.content, end="", flush=True)
    print()  # newline at end

if __name__ == "__main__":
    asyncio.run(stream_response("Explain the difference between a list and a tuple in Python. Keep it under 100 words."))

Run python streaming.py. Tokens should appear smoothly. For a FastAPI endpoint, return a StreamingResponse that yields chunk.content from the same async generator.

Step 5: Add tool calling with structured output

Llama 3.3 70B supports function calling. Define your schema with Pydantic, bind tools to the model, and parse the result.

# tools.py
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.utils.function_calling import convert_to_openai_tool
from llm_setup import llm

class GetWeather(BaseModel):
    """Get current weather for a location."""
    location: str = Field(..., description="City and state, e.g., 'San Francisco, CA'")
    unit: Literal["celsius", "fahrenheit"] = "fahrenheit"

class SearchDocs(BaseModel):
    """Search internal documentation."""
    query: str = Field(..., description="Search query")
    top_k: int = Field(default=5, ge=1, le=20)

tools = [convert_to_openai_tool(t) for t in (GetWeather, SearchDocs)]
llm_with_tools = llm.bind_tools(tools)

# Test
messages = [
    ("system", "You have access to weather and documentation search. Use tools when appropriate."),
    ("human", "What's the weather in Austin, TX right now?"),
]

response = llm_with_tools.invoke(messages)
print(response.tool_calls)

Output example:

[{'name': 'GetWeather', 'args': {'location': 'Austin, TX', 'unit': 'fahrenheit'}, 'id': 'call_abc123'}]

Execute the tool call (you implement the actual function):

# tool_execution.py
import json
from tools import llm_with_tools, GetWeather, SearchDocs

def get_weather(location: str, unit: str) -> dict:
    # Replace with real API call
    return {"location": location, "temperature": 72, "unit": unit, "condition": "sunny"}

def search_docs(query: str, top_k: int) -> list:
    return [{"title": "Doc 1", "snippet": "..."}, {"title": "Doc 2", "snippet": "..."}]

TOOL_MAP = {
    "GetWeather": get_weather,
    "SearchDocs": search_docs,
}

def run_agent(user_query: str):
    messages = [
        ("system", "You have access to weather and documentation search. Use tools when appropriate."),
        ("human", user_query),
    ]
    response = llm_with_tools.invoke(messages)
    
    for tool_call in response.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        result = TOOL_MAP[tool_name](**tool_args)
        messages.append(response)
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call["id"],
            "content": json.dumps(result),
        })
    
    final = llm_with_tools.invoke(messages)
    return final.content

if __name__ == "__main__":
    print(run_agent("What's the weather in Austin, TX?"))

Step 6: Structure output with JSON mode

For deterministic parsing, use JSON mode with a Pydantic schema. This avoids brittle regex extraction.

# structured_output.py
from pydantic import BaseModel, Field
from typing import List
from llm_setup import llm

class CodeReview(BaseModel):
    file_path: str
    issues: List[str] = Field(default_factory=list)
    suggestions: List[str] = Field(default_factory=list)
    severity: Literal["low", "medium", "high", "critical"]
    approved: bool

structured_llm = llm.with_structured_output(CodeReview, method="json_mode")

code_snippet = """
def get_user(id):
    query = f"SELECT * FROM users WHERE id = {id}"
    return db.execute(query)
"""

prompt = f"Review this code for security issues:\n{code_snippet}"
result = structured_llm.invoke(prompt)
print(result.model_dump_json(indent=2))

Sample output:

{
  "file_path": "user_lookup.py",
  "issues": ["SQL injection via f-string interpolation"],
  "suggestions": ["Use parameterized queries: cursor.execute('SELECT * FROM users WHERE id = %s', (id,))"],
  "severity": "critical",
  "approved": false
}

Step 7: Implement retries, timeouts, and fallback logic

Production code needs resilience. n4n.ai automatically falls back across providers when one is rate-limited or degraded, but you should still handle transient network errors and enforce deadlines at the client layer.

# resilient_llm.py
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableConfig
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
import httpx

load_dotenv()

class ResilientChatOpenAI(ChatOpenAI):
    def __init__(self, **kwargs):
        # Default timeouts: connect 10s, read 120s for long generations
        kwargs.setdefault("timeout", httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=5.0))
        kwargs.setdefault("max_retries", 0)  # we handle retries ourselves
        super().__init__(**kwargs)

    @retry(
        wait=wait_exponential_jitter(initial=1, max=30),
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError)),
        reraise=True,
    )
    def invoke(self, input, config=None, **kwargs):
        return super().invoke(input, config, **kwargs)

    @retry(
        wait=wait_exponential_jitter(initial=1, max=30),
        stop=stop_after_attempt(3),
        retry=retry_if_exception_type((httpx.TimeoutException, httpx.ConnectError, httpx.RemoteProtocolError)),
        reraise=True,
    )
    async def ainvoke(self, input, config=None, **kwargs):
        return await super().ainvoke(input, config, **kwargs)

# Usage
resilient_llm = ResilientChatOpenAI(
    model=os.getenv("LLAMA_MODEL"),
    api_key=os.getenv("N4N_API_KEY"),
    base_url=os.getenv("N4N_BASE_URL"),
    temperature=0.2,
    max_tokens=4096,
)

Add tenacity to your dependencies: pip install tenacity. The max_retries=0 on the base class prevents double-retrying; the decorator handles it with exponential backoff and jitter.

Step 8: Add observability — token counting and latency

You need per-request token usage for cost tracking and latency percentiles for SLOs. LangChain’s get_openai_callback works with any OpenAI-compatible endpoint.

# observability.py
from contextlib import contextmanager
import time
from langchain_community.callbacks.manager import get_openai_callback
from resilient_llm import resilient_llm

@contextmanager
def track_llm_call(operation: str):
    start = time.perf_counter()
    with get_openai_callback() as cb:
        yield cb
    elapsed = time.perf_counter() - start
    print(f"[{operation}] latency={elapsed:.3f}s "
          f"prompt_tokens={cb.prompt_tokens} "
          f"completion_tokens={cb.completion_tokens} "
          f"total_tokens={cb.total_tokens} "
          f"estimated_cost_usd=${cb.total_cost:.6f}")

# Example
with track_llm_call("code_review") as cb:
    result = resilient_llm.invoke("Write a 50-word haiku about Kubernetes.")
    print(result.content[:80] + "...")

Output:

[code_review] latency=1.847s prompt_tokens=42 completion_tokens=38 total_tokens=80 estimated_cost_usd=$0.000240

For production, push these metrics to Prometheus, Datadog, or your observability stack instead of printing.

Step 9: Handle context window limits gracefully

Llama 3.3 70B has a 128k context window. Long conversations or RAG pipelines can exceed this. Implement a token-aware trimmer.

# context_management.py
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_core.utils import get_tokenizer
from typing import List

# Use cl100k_base as a close approximation for Llama 3 tokenization
tokenizer = get_tokenizer("cl100k_base")

MAX_CONTEXT_TOKENS = 120_000  # leave headroom for response
RESERVED_RESPONSE_TOKENS = 4096

def count_tokens(messages: List[BaseMessage]) -> int:
    total = 0
    for m in messages:
        total += len(tokenizer.encode(m.content))
        total += 4  # rough overhead per message (role, formatting)
    return total

def trim_messages(messages: List[BaseMessage], max_tokens: int = MAX_CONTEXT_TOKENS) -> List[BaseMessage]:
    """Drop oldest non-system messages until under limit."""
    if count_tokens(messages) <= max_tokens:
        return messages
    
    system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
    other_msgs = [m for m in messages if not isinstance(m, SystemMessage)]
    
    while other_msgs and count_tokens(system_msgs + other_msgs) > max_tokens:
        other_msgs.pop(0)  # drop oldest
    
    return system_msgs + other_msgs

# Usage in a conversation loop
messages = [
    SystemMessage(content="You are a helpful coding assistant."),
    HumanMessage(content="Write a Python class for a binary search tree."),
    AIMessage(content="Here's a BST implementation..."),
    # ... many more turns ...
]

messages = trim_messages(messages)
response = resilient_llm.invoke(messages)

For precise token counts, use the model’s actual tokenizer via transformers or tiktoken with the correct encoding. The approximation above is safe for guardrails.

Step 10: Deploy behind a feature flag

Wrap the integration in a feature flag so you can roll back instantly if the model behaves unexpectedly in production.

# feature_flagged_llm.py
import os
from resilient_llm import resilient_llm
from llm_setup import llm as fallback_llm  # could be a smaller/cheaper model

USE_LLAMA_3_3_70B = os.getenv("USE_LLAMA_3_3_70B", "false").lower() == "true"

def get_llm():
    if USE_LLAMA_3_3_70B:
        return resilient_llm
    return fallback_llm

# In your application code
llm = get_llm()
response = llm.invoke("Summarize the attached PR diff in three bullet points.")

Toggle USE_LLAMA_3_3_70B=true in your deployment config to enable. This pattern also lets you A/B test against other models routed through the same endpoint.

Verification checklist

Before marking the integration done, run through these:

1. Basic invocation — python test_basic.py returns a coherent response.
2. Streaming — python streaming.py prints tokens incrementally without blocking.
3. Tool calling — python tool_execution.py executes the weather tool and returns a final answer incorporating the result.
4. Structured output — python structured_output.py emits valid JSON matching your Pydantic schema.
5. Resilience — Simulate a network partition (e.g., tc qdisc add dev lo root netem loss 50%) and verify retries succeed with exponential backoff.
6. Observability — Confirm token counts and latency appear in your metrics pipeline.
7. Context trimming — Feed a 150k-token conversation; verify the trimmer drops oldest turns and the request succeeds.
8. Feature flag — Flip USE_LLAMA_3_3_70B=false; verify fallback model serves traffic without code changes.

What to tune next

• Temperature and top_p: For coding tasks, temperature=0.1, top_p=0.95 reduces hallucination. For creative writing, raise to 0.7.
• Max tokens: Set per-use-case. Code generation often needs 4k–8k; classification needs 100.
• System prompt: Invest in a strong system prompt with few-shot examples for your domain. It matters more than temperature.
• Batch requests: If you have high-throughput async workloads, use abatch with a semaphore to control concurrency and avoid 429s.

You now have a production-grade Llama 3.3 70B integration in LangChain. The same pattern applies to any model routed through the endpoint — swap the model ID and you’re running Mixtral, Qwen, or the next release without changing application code.