tati-langchain

A framework-agnostic AI/LangChain engine you can pip install into any Python project (Django, FastAPI, Flask, a script, …).

Covers the pieces every AI app ends up re-building:

Every Django/ORM/settings dependency from the source project has been swapped for plain dataclasses and explicit function arguments.

Install

Published on PyPI.

pip install tati-langchain

# + AWS Bedrock support
pip install "tati-langchain[bedrock]"

From a consuming project (requirements.in)

tati-langchain>=0.3.1
# or with Bedrock:
# tati-langchain[bedrock]>=0.3.1

Then:

pip install -r requirements.in

Set credentials the normal LangChain way:

• OpenAI → OPENAI_API_KEY
• Anthropic → ANTHROPIC_API_KEY
• Bedrock → standard AWS credentials (AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_REGION, or an instance role)

1. Pick a provider and build a model

from decimal import Decimal
from tati_langchain import ModelSpec, ProviderStack, Provider, build_chat_model

# --- OpenAI ---
openai_stack = ProviderStack(
    provider=Provider.OPENAI,
    display_name="OpenAI",
    chat_model=ModelSpec(
        provider=Provider.OPENAI,
        name="gpt-5.4-mini",
        supports_vision=True,
        supports_tools=True,
        max_output_tokens=4096,
        input_cost_per_1m_tokens=Decimal("0.15"),
        output_cost_per_1m_tokens=Decimal("0.60"),
    ),
    supports_web_search=True,
)
openai_bundle = build_chat_model(openai_stack)

# --- Anthropic ---
anthropic_stack = ProviderStack(
    provider=Provider.ANTHROPIC,
    display_name="Anthropic",
    chat_model=ModelSpec(
        provider=Provider.ANTHROPIC,
        name="claude-haiku-4-5",
        max_output_tokens=4096,
        input_cost_per_1m_tokens=Decimal("0.80"),
        output_cost_per_1m_tokens=Decimal("4.00"),
    ),
    supports_web_search=True,
)
anthropic_bundle = build_chat_model(anthropic_stack)

# --- AWS Bedrock (Converse API — best for tool calling) ---
# requires: pip install "tati-langchain[bedrock]"
bedrock_stack = ProviderStack(
    provider=Provider.BEDROCK_CONVERSE,
    display_name="Bedrock",
    chat_model=ModelSpec(
        provider=Provider.BEDROCK_CONVERSE,
        name="anthropic.claude-3-5-sonnet-20241022-v2:0",
        max_output_tokens=4096,
        extra_params={"region_name": "eu-west-1"},  # forwarded to ChatBedrockConverse
    ),
    supports_web_search=False,  # no native Bedrock web-search tool in this package
)
bedrock_bundle = build_chat_model(bedrock_stack, include_default_tools=False)

build_chat_model returns a ChatModelBundle:

• bundle.chat_llm — model with tools bound (use with run_tool_loop)
• bundle.raw_llm — unbound model (use with generate_text / generate_structured)
• bundle.tools_by_name — local tools the agent loop can execute

2. Text generation

from langchain_core.messages import HumanMessage, SystemMessage
from tati_langchain import generate_text

result = generate_text(
    openai_bundle.raw_llm,
    [
        SystemMessage("You are a concise assistant."),
        HumanMessage("Explain vector databases in two sentences."),
    ],
    model=openai_stack.chat_model,  # optional — enables result.cost
)
print(result.text)
print(result.usage)          # {"input_tokens", "output_tokens", "cached_input_tokens"}
print(result.cost.total_cost if result.cost else None)

3. Long-form writing & research

Long-form = high max_output_tokens. Research = turn on native web search (OpenAI / Anthropic) and ask the model to cite sources.

from langchain_core.messages import HumanMessage, SystemMessage
from tati_langchain import ModelSpec, Provider, ProviderStack, build_chat_model, run_tool_loop

research_stack = ProviderStack(
    provider=Provider.OPENAI,
    display_name="Research",
    chat_model=ModelSpec(
        provider=Provider.OPENAI,
        name="gpt-5.4",
        max_output_tokens=16000,  # long-form headroom
    ),
    supports_web_search=True,     # binds the provider-native web_search tool
)
bundle = build_chat_model(research_stack)

messages = [
    SystemMessage(
        "You are a research analyst. Use web search. Write a structured brief "
        "with a summary, key findings, and cited sources."
    ),
    HumanMessage("What changed in EU AI Act enforcement in the last 6 months?"),
]
result = run_tool_loop(bundle.chat_llm, messages, bundle.tools_by_name)
print(result.ai_message.content)

4. Image generation

Built-in generate_image tool (OpenAI Images API). Works even on an Anthropic / Bedrock chat stack if you point image_model at an OpenAI image model.

from decimal import Decimal
from langchain_core.messages import HumanMessage
from tati_langchain import (
    ImageModelSpec, ModelSpec, Provider, ProviderStack,
    build_chat_model, run_tool_loop, calculate_image_cost,
)

stack = ProviderStack(
    provider=Provider.OPENAI,
    display_name="Creative",
    chat_model=ModelSpec(provider=Provider.OPENAI, name="gpt-5.4-mini"),
    image_model=ImageModelSpec(
        provider=Provider.OPENAI,
        name="gpt-image-1-mini",
        text_input_cost_per_1m=Decimal("5.00"),
        image_output_cost_per_1m=Decimal("40.00"),
    ),
)
bundle = build_chat_model(stack)
result = run_tool_loop(
    bundle.chat_llm,
    [HumanMessage("Draw a red fox wearing sunglasses")],
    bundle.tools_by_name,
    on_progress=print,  # optional: "🎨 Image generation triggered..."
)

for att in result.attachments:
    open("fox.png", "wb").write(att.data)

for usage in result.image_usages:
    print(calculate_image_cost(model=stack.image_model, **usage["tokens"]))

5. Cost extraction (what a run actually cost)

from tati_langchain import extract_usage, calculate_message_cost, cost_for_agent_result

# Plain text turn
usage = extract_usage(result.ai_message)
breakdown = calculate_message_cost(model=stack.chat_model, **usage)
print(breakdown.total_cost, breakdown.currency)

# Full agent turn (chat tokens + any image tool usages)
message_cost, image_costs = cost_for_agent_result(
    ai_message=result.ai_message,
    model=stack.chat_model,
    image_usages=result.image_usages,
)
print(message_cost.total_cost, [c.total_cost for c in image_costs])

Nothing is persisted — you decide whether that becomes a DB row, a log line, or a metrics counter.

6. Structured outputs with Pydantic

from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage
from tati_langchain import generate_structured, calculate_message_cost

class BookRec(BaseModel):
    title: str
    author: str
    reason: str = Field(description="One-sentence why this fits")

structured = generate_structured(
    openai_bundle.raw_llm,
    [HumanMessage("Recommend one sci-fi book for a beginner.")],
    BookRec,
)
print(structured.parsed.title, structured.parsed.author)
print(structured.usage)
if structured.raw_message is not None:
    print(calculate_message_cost(model=openai_stack.chat_model, **structured.usage))

Or bind once and reuse:

from tati_langchain import with_structured_output

llm = with_structured_output(openai_bundle.raw_llm, BookRec)
rec = llm.invoke([HumanMessage("Recommend a mystery novel.")])

7. Agentic design + custom tools

from langchain_core.messages import HumanMessage
from tati_langchain import define_tool, build_chat_model, run_tool_loop, bind_extra_tools

@define_tool
def lookup_order(order_id: str) -> str:
    """Look up an order by id and return its status."""
    return f"Order {order_id}: shipped"

# Option A — pass extra tools at build time
bundle = build_chat_model(openai_stack, extra_tools=[lookup_order])

# Option B — rebind onto an existing bundle
bundle = bind_extra_tools(bundle, [lookup_order])

result = run_tool_loop(
    bundle.chat_llm,
    [HumanMessage("Where is order A-100?")],
    bundle.tools_by_name,
)
print(result.ai_message.content)

run_tool_loop is provider-agnostic: it invokes the model, executes any local tool calls registered in tools_by_name, feeds results back, and stops after a small iteration cap (or when a tool signals forced_reply / limit_reached). Provider-native tools (e.g. web search) never appear in tool_calls — the provider resolves them server-side.

Tool with an explicit Pydantic args schema

from pydantic import BaseModel, Field
from tati_langchain import define_tool

class SearchArgs(BaseModel):
    query: str
    limit: int = Field(default=5, ge=1, le=20)

@define_tool(args_schema=SearchArgs)
def search_docs(query: str, limit: int = 5) -> str:
    """Search the internal docs corpus."""
    return f"top {limit} hits for {query!r}"

8. Document generation

generate_document degrades gracefully (returns an "unavailable" message to the model, doesn't raise) until the optional tati-docgen package is also installed — at which point it starts working with no code change.

Design principles

• You own persistence, config, and "what's active." This package never reads a global settings object and never writes to a database.
• Pure cost math, no side effects. Cost helpers return dataclasses; you decide how to store them.
• No messaging dependency. Tool attachments come back as this package's own ToolAttachment — map to WhatsApp/email/etc. at the call site.

What's not in this package (by design)

• Model-catalog / "which stack is active" storage
• Conversation history storage
• Free-trial / usage-limit gating (tools may still signal limit_reached)
• Sending replies to WhatsApp/email (see tati-whatsapp)
• i18n for progress strings — override via progress_text_builders=

License

Proprietary — internal use only within Tati Software Pty Ltd. See LICENSE.