How to Use Llama 3.2 70B and LangChain 0.3 to Generate Smart Contracts for Solidity 0.8.20

import os
import sys
import subprocess
import time
from packaging import version
from ollama import Client

def check_prerequisites():
    """Verify all required tools are installed and meet version requirements."""
    errors = []

    # Check Python version >= 3.10
    if sys.version_info < (3, 10):
        errors.append(f"Python 3.10+ required, found {sys.version}")

    # Check Ollama installation and version
    try:
        ollama_version = subprocess.run(
            ["ollama", "version"], 
            capture_output=True, 
            text=True, 
            check=True
        ).stdout.strip().split()[-1]
        if version.parse(ollama_version) < version.parse("0.5.0"):
            errors.append(f"Ollama 0.5.0+ required, found {ollama_version}")
    except FileNotFoundError:
        errors.append("Ollama not installed. Install from https://ollama.com")
    except subprocess.CalledProcessError as e:
        errors.append(f"Failed to check Ollama version: {e.stderr}")

    # Check Llama 3.2 70B model is pulled
    try:
        client = Client()
        models = [m["name"] for m in client.list()["models"]]
        if "llama3.2:70b" not in models:
            errors.append("Llama 3.2 70B not pulled. Run: ollama pull llama3.2:70b")
    except Exception as e:
        errors.append(f"Failed to connect to Ollama: {str(e)}")

    # Check Solidity compiler version 0.8.20
    try:
        solc_version = subprocess.run(
            ["solc", "--version"], 
            capture_output=True, 
            text=True, 
            check=True
        ).stdout.strip().split("\n")[1].split(" ")[1]
        if solc_version != "0.8.20":
            errors.append(f"solc 0.8.20 required, found {solc_version}")
    except FileNotFoundError:
        errors.append("solc not installed. Install from https://soliditylang.org")

    # Check LangChain version 0.3+
    try:
        import langchain
        if version.parse(langchain.__version__) < version.parse("0.3.0"):
            errors.append(f"LangChain 0.3.0+ required, found {langchain.__version__}")
    except ImportError:
        errors.append("LangChain not installed. Run: pip install langchain==0.3.2")

    return errors

def install_dependencies():
    """Install required Python packages."""
    reqs = [
        "langchain==0.3.2",
        "langchain-community==0.3.1",
        "ollama==0.5.0",
        "packaging==24.1",
        "py-solc-x==2.0.2"
    ]
    for req in reqs:
        print(f"Installing {req}...")
        subprocess.run([sys.executable, "-m", "pip", "install", req], check=True)
    print("All dependencies installed.")

if __name__ == "__main__":
    print("Checking prerequisites for Llama 3.2 70B + LangChain 0.3 Solidity pipeline...")
    start = time.time()
    errors = check_prerequisites()

    if errors:
        print("\n❌ Prerequisite errors found:")
        for err in errors:
            print(f"  - {err}")
        print("\nInstalling dependencies...")
        try:
            install_dependencies()
            print("\n✅ Dependencies installed. Re-run this script to verify prerequisites.")
        except Exception as e:
            print(f"\n❌ Failed to install dependencies: {str(e)}")
            sys.exit(1)
    else:
        print("\n✅ All prerequisites met. Proceeding to pipeline setup.")
        print(f"Prerequisite check completed in {time.time() - start:.2f} seconds.")

import os
import json
import time
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.output_parsers.solidity import SolidityOutputParser
from solidity_compiler import validate_solidity  # Local validator from src/validators

# Few-shot examples for ERC20 token generation (Solidity 0.8.20)
erc20_examples = [
    {
        "input": "ERC20 token with name 'TestToken', symbol 'TT', 1M total supply, 1% transfer fee to 0x123...",
        "output": """// SPDX-License-Identifier: MIT
pragma solidity 0.8.20;

import "@openzeppelin/contracts/token/ERC20/ERC20.sol";
import "@openzeppelin/contracts/access/Ownable.sol";

contract TestToken is ERC20, Ownable {
    address public treasury;
    uint256 public constant FEE_BASIS_POINTS = 100; // 1% = 100 basis points
    uint256 public constant BASIS_POINTS_DENOMINATOR = 10000;

    constructor(address _treasury) ERC20("TestToken", "TT") Ownable(msg.sender) {
        treasury = _treasury;
        _mint(msg.sender, 1_000_000 * 10 ** decimals());
    }

    function _update(address from, address to, uint256 value) internal override {
        if (from != address(0) && to != address(0)) {
            uint256 fee = (value * FEE_BASIS_POINTS) / BASIS_POINTS_DENOMINATOR;
            super._update(from, treasury, fee);
            super._update(from, to, value - fee);
        } else {
            super._update(from, to, value);
        }
    }

    function setTreasury(address _treasury) external onlyOwner {
        treasury = _treasury;
    }
}
"""
    },
    {
        "input": "ERC20 token with mintable and burnable functionality, 2% transfer fee",
        "output": """// SPDX-License-Identifier: MIT
pragma solidity 0.8.20;

import "@openzeppelin/contracts/token/ERC20/extensions/ERC20Burnable.sol";
import "@openzeppelin/contracts/access/Ownable.sol";

contract MintableBurnableToken is ERC20Burnable, Ownable {
    uint256 public transferFee;
    address public feeRecipient;

    constructor(string memory name, string memory symbol, uint256 _transferFee, address _feeRecipient) 
        ERC20(name, symbol) Ownable(msg.sender) {
        transferFee = _transferFee;
        feeRecipient = _feeRecipient;
    }

    function mint(address to, uint256 amount) external onlyOwner {
        _mint(to, amount);
    }

    function _update(address from, address to, uint256 value) internal override {
        if (from != address(0) && to != address(0) && transferFee > 0) {
            uint256 fee = (value * transferFee) / 10000;
            super._update(from, feeRecipient, fee);
            super._update(from, to, value - fee);
        } else {
            super._update(from, to, value);
        }
    }
}
"""
    }
]

def build_solidity_chain():
    """Build LangChain pipeline for Solidity 0.8.20 generation."""
    # Initialize Llama 3.2 70B via Ollama with temperature 0.1 for deterministic output
    llm = ChatOllama(
        model="llama3.2:70b",
        temperature=0.1,
        top_p=0.95,
        num_ctx=8192  # Support long Solidity contracts
    )

    # Few-shot prompt template
    example_prompt = ChatPromptTemplate.from_messages([
        ("human", "{input}"),
        ("ai", "{output}")
    ])
    few_shot_prompt = FewShotChatMessagePromptTemplate(
        examples=erc20_examples,
        example_prompt=example_prompt
    )

    # Main prompt template with strict Solidity 0.8.20 rules
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a senior Solidity developer specializing in Solidity 0.8.20 and OpenZeppelin v5.0.0.
        Generate only valid Solidity 0.8.20 code, no explanatory text.
        Follow these rules:
        1. Use pragma solidity 0.8.20 exactly
        2. Import OpenZeppelin contracts from @openzeppelin/contracts@5.0.0
        3. Use 0x-prefixed checksum addresses for constants
        4. Include SPDX license identifier at top
        5. Override _update for ERC20 transfer hooks (Solidity 0.8.20+)
        6. Use underscore separators for large numbers (e.g., 1_000_000)
        """),
        few_shot_prompt,
        ("human", "{requirement}")
    ])

    # Output parser to validate Solidity syntax and 0.8.20 compliance
    output_parser = SolidityOutputParser(
        required_pragma="0.8.20",
        allowed_imports=["@openzeppelin/contracts@5.0.0"]
    )

    # Build chain: prompt -> llm -> output parser
    chain = prompt | llm | StrOutputParser() | output_parser
    return chain

def generate_contract(requirement: str):
    """Generate a Solidity contract from natural language requirement."""
    chain = build_solidity_chain()
    start = time.time()
    try:
        result = chain.invoke({"requirement": requirement})
        # Validate compiled code with solc 0.8.20
        is_valid, errors = validate_solidity(result["code"])
        if not is_valid:
            return {"success": False, "errors": errors, "code": None}
        return {
            "success": True,
            "code": result["code"],
            "generation_time": time.time() - start
        }
    except Exception as e:
        return {"success": False, "errors": [str(e)], "code": None}

if __name__ == "__main__":
    test_requirement = "ERC20 token named 'DemoToken', symbol 'DT', 500k supply, 0.5% transfer fee to 0x742d35Cc6634C0532925a3b8D0fC05bb79D9A3b4"
    print(f"Generating contract for: {test_requirement}")
    result = generate_contract(test_requirement)
    if result["success"]:
        print(f"\n✅ Contract generated in {result['generation_time']:.2f}s:")
        print(result["code"])
    else:
        print(f"\n❌ Generation failed: {result['errors']}")

import os
import subprocess
import json
from pathlib import Path
from typing import Dict, List, Tuple

class SolidityValidator:
    """Validate Solidity 0.8.20 contracts and generate Foundry tests."""
    def __init__(self, solc_version: str = "0.8.20"):
        self.solc_version = solc_version
        self.openzeppelin_version = "5.0.0"
        self.foundry_binary = "forge"

    def compile_contract(self, contract_code: str, contract_name: str) -> Tuple[bool, List[str]]:
        """Compile contract with solc 0.8.20 and return success/errors."""
        temp_dir = Path("temp_contracts")
        temp_dir.mkdir(exist_ok=True)
        contract_path = temp_dir / f"{contract_name}.sol"

        try:
            # Write contract to temp file
            contract_path.write_text(contract_code)

            # Compile with solc 0.8.20
            result = subprocess.run(
                [
                    "solc", f"--{self.solc_version}",
                    "--optimize", "--optimize-runs", "200",
                    "--abi", "--bin",
                    str(contract_path),
                    "-o", str(temp_dir / "build")
                ],
                capture_output=True,
                text=True,
                check=True
            )
            return True, []
        except subprocess.CalledProcessError as e:
            errors = [line for line in e.stderr.split("\n") if "Error" in line]
            return False, errors
        except Exception as e:
            return False, [str(e)]
        finally:
            # Cleanup temp files
            if contract_path.exists():
                contract_path.unlink()

    def check_openzeppelin_compliance(self, contract_code: str) -> Tuple[bool, List[str]]:
        """Check if contract uses OpenZeppelin 5.0.0 correctly."""
        errors = []
        required_imports = [
            "@openzeppelin/contracts/token/ERC20/ERC20.sol",
            "@openzeppelin/contracts/access/Ownable.sol"
        ]

        for req in required_imports:
            if req not in contract_code and "ERC20" in contract_code:
                errors.append(f"Missing required OpenZeppelin import: {req}")

        # Check for deprecated Solidity 0.8.20 patterns
        if "pragma solidity ^0.8.0" in contract_code:
            errors.append("Use exact pragma solidity 0.8.20, not ^0.8.0")
        if "transfer(" in contract_code and "super._update" not in contract_code:
            errors.append("ERC20 transfers should override _update for Solidity 0.8.20+")

        return len(errors) == 0, errors

    def generate_foundry_tests(self, contract_code: str, contract_name: str) -> str:
        """Generate basic Foundry unit tests for the contract."""
        # Extract contract functions (simplified parser)
        functions = []
        for line in contract_code.split("\n"):
            if line.strip().startswith("function "):
                func_name = line.split("function ")[1].split("(")[0]
                functions.append(func_name)

        # Generate test template
        test_code = f"""// SPDX-License-Identifier: MIT
pragma solidity 0.8.20;

import "forge-std/Test.sol";
import "../src/{contract_name}.sol";

contract {contract_name}Test is Test {{
    {contract_name} public token;
    address public owner = address(this);
    address public treasury = address(0x123);

    function setUp() public {{
        token = new {contract_name}(treasury);
    }}

    function test_initial_supply() public {{
        assertEq(token.totalSupply(), 500_000 * 10 ** token.decimals());
    }}

    function test_transfer_fee() public {{
        address alice = address(0x456);
        uint256 amount = 1000 * 10 ** token.decimals();
        token.transfer(alice, amount);
        uint256 fee = (amount * 50) / 10000; // 0.5% fee
        assertEq(token.balanceOf(treasury), fee);
        assertEq(token.balanceOf(alice), amount - fee);
    }}
}}
"""
        # Add test functions for each extracted function
        for func in functions:
            if func not in ["setUp", "test_initial_supply", "test_transfer_fee"]:
                test_code += f"""
    function test_{func}() public {{
        // TODO: Implement test for {func}
    }}
"""
        return test_code

    def run_foundry_tests(self, contract_code: str, contract_name: str) -> Tuple[bool, str]:
        """Run Foundry tests for the contract."""
        temp_dir = Path("temp_contracts")
        src_dir = temp_dir / "src"
        test_dir = temp_dir / "test"
        src_dir.mkdir(parents=True, exist_ok=True)
        test_dir.mkdir(exist_ok=True)

        # Write contract and tests
        (src_dir / f"{contract_name}.sol").write_text(contract_code)
        test_code = self.generate_foundry_tests(contract_code, contract_name)
        (test_dir / f"{contract_name}.t.sol").write_text(test_code)

        # Run forge test
        try:
            result = subprocess.run(
                ["forge", "test", "-vvv", "--root", str(temp_dir)],
                capture_output=True,
                text=True,
                check=True
            )
            return True, result.stdout
        except subprocess.CalledProcessError as e:
            return False, e.stderr
        finally:
            # Cleanup
            import shutil
            if temp_dir.exists():
                shutil.rmtree(temp_dir)

if __name__ == "__main__":
    validator = SolidityValidator()
    # Load sample contract from previous step
    sample_contract = """// SPDX-License-Identifier: MIT
pragma solidity 0.8.20;

import "@openzeppelin/contracts/token/ERC20/ERC20.sol";
import "@openzeppelin/contracts/access/Ownable.sol";

contract DemoToken is ERC20, Ownable {
    address public treasury;
    uint256 public constant FEE_BASIS_POINTS = 50; // 0.5%
    uint256 public constant BASIS_POINTS_DENOMINATOR = 10000;

    constructor(address _treasury) ERC20("DemoToken", "DT") Ownable(msg.sender) {
        treasury = _treasury;
        _mint(msg.sender, 500_000 * 10 ** decimals());
    }

    function _update(address from, address to, uint256 value) internal override {
        if (from != address(0) && to != address(0)) {
            uint256 fee = (value * FEE_BASIS_POINTS) / BASIS_POINTS_DENOMINATOR;
            super._update(from, treasury, fee);
            super._update(from, to, value - fee);
        } else {
            super._update(from, to, value);
        }
    }
}
"""
    print("Compiling contract...")
    success, errors = validator.compile_contract(sample_contract, "DemoToken")
    if success:
        print("✅ Contract compiled successfully.")
        print("Checking OpenZeppelin compliance...")
        oz_success, oz_errors = validator.check_openzeppelin_compliance(sample_contract)
        if oz_success:
            print("✅ OpenZeppelin compliance passed.")
            print("Generating and running Foundry tests...")
            test_success, test_output = validator.run_foundry_tests(sample_contract, "DemoToken")
            if test_success:
                print("✅ All Foundry tests passed.")
            else:
                print(f"❌ Foundry tests failed: {test_output}")
        else:
            print(f"❌ OpenZeppelin compliance failed: {oz_errors}")
    else:
        print(f"❌ Compilation failed: {errors}")

from langchain_community.chat_models import ChatOllama
from langchain.cache import RedisCache
from langchain.globals import set_llm_cache

set_llm_cache(RedisCache(host="localhost", port=6379, db=0))
llm = ChatOllama(model="llama3.2:70b", cache=True)

from langchain_core.output_parsers import BaseOutputParser

class Solidity08OutputParser(BaseOutputParser):
    def parse(self, text: str) -> dict:
        errors = []
        if "pragma solidity 0.8.20;" not in text:
            errors.append("Missing exact pragma solidity 0.8.20")
        if "@openzeppelin/contracts@5.0.0" not in text and "openzeppelin" in text.lower():
            errors.append("OpenZeppelin imports must be pinned to 5.0.0")
        if "_beforeTokenTransfer" in text:
            errors.append("Use _update instead of _beforeTokenTransfer for Solidity 0.8.20")
        return {"code": text, "errors": errors} if errors else {"code": text}

FROM llama3.2:70b

# Use 4-bit quantization (Q4_K_M)
PARAMETER quantize q4_k_m

# Set context length to 8192 for long contracts
PARAMETER num_ctx 8192

# Lower temperature for deterministic Solidity output
PARAMETER temperature 0.1

llama-solidity-generator/
├── src/
│   ├── chain.py               # Main LangChain pipeline
│   ├── validators/
│   │   ├── solidity_compiler.py  # solc 0.8.20 validation
│   │   └── audit_checker.py      # OpenZeppelin audit rules
│   ├── prompts/
│   │   ├── base_prompt.py        # Few-shot Solidity prompts
│   │   └── erc20_prompt.py       # ERC20-specific templates
│   └── tests/
│       ├── test_foundry.py        # Foundry test generator
│       └── test_contracts.py      # Pipeline integration tests
├── requirements.txt             # LangChain 0.3.2, ollama 0.5.0, etc.
├── ollama_modelfile             # 4-bit quantized Llama 3.2 70B config
└── README.md                   # Setup and usage instructions