Migrating from raw OpenAI function calling to LangChain tools reduces boilerplate, centralizes schema management, and makes it easier to swap models or add observability. The OpenAI SDK approach works fine for prototypes, but production systems benefit from LangChain’s abstraction layer — especially when you need consistent tool interfaces across multiple providers. This walkthrough takes you from a working OpenAI functions implementation to equivalent LangChain tool definitions with minimal friction.
Step 1: inventory your existing function definitions
Start by extracting every function schema from your current codebase. Raw OpenAI function calling typically scatters JSON schemas across route handlers, service classes, or prompt templates. Collect them into a single inventory file so you can map each one to a LangChain tool.
# inventory.py — run this against your codebase to extract schemas
import ast
import json
import sys
from pathlib import Path
def extract_openai_functions(filepath: Path) -> list[dict]:
"""Parse a Python file and return all dicts passed as `functions=` to openai.ChatCompletion.create."""
tree = ast.parse(filepath.read_text())
functions = []
for node in ast.walk(tree):
if isinstance(node, ast.Call):
# Look for openai.ChatCompletion.create(functions=[...])
if (isinstance(node.func, ast.Attribute)
and node.func.attr == "create"
and isinstance(node.func.value, ast.Attribute)
and node.func.value.attr == "ChatCompletion"):
for kw in node.keywords:
if kw.arg == "functions" and isinstance(kw.value, ast.List):
for elt in kw.value.elts:
if isinstance(elt, ast.Dict):
func_dict = {}
for k, v in zip(elt.keys, elt.values):
if isinstance(k, ast.Constant):
func_dict[k.value] = ast.literal_eval(v)
functions.append(func_dict)
return functions
if __name__ == "__main__":
all_functions = []
for py_file in Path(".").rglob("*.py"):
all_functions.extend(extract_openai_functions(py_file))
print(json.dumps(all_functions, indent=2))Run this script and save the output. You’ll need each function’s name, description, and parameters (JSON Schema) for the next step.
Verify: The printed JSON should match every function you currently register with the OpenAI SDK. Count them — this is your migration target.
Step 2: convert each schema to a LangChain tool
LangChain tools wrap a callable with a name, description, and argument schema. The cleanest path is StructuredTool.from_function, which infers the schema from type hints and docstrings. For each OpenAI function, create a corresponding Python function with proper annotations.
# tools/weather.py
from typing import Literal
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
class GetWeatherArgs(BaseModel):
location: str = Field(..., description="City and state, e.g. 'San Francisco, CA'")
unit: Literal["celsius", "fahrenheit"] = Field("fahrenheit", description="Temperature unit")
def get_weather(location: str, unit: str = "fahrenheit") -> dict:
"""Fetch current weather for a location."""
# Your existing implementation here — keep the business logic unchanged
import requests
resp = requests.get(
"https://api.weather.example.com/current",
params={"q": location, "units": "imperial" if unit == "fahrenheit" else "metric"},
timeout=5
)
resp.raise_for_status()
return resp.json()
get_weather_tool = StructuredTool.from_function(
func=get_weather,
name="get_weather",
description="Get current weather conditions for a specified location",
args_schema=GetWeatherArgs,
return_direct=False, # set True only if the tool result should go straight to the user
)Repeat for every function in your inventory. Key points:
- Keep the original function name — the LLM sees this identifier.
- Mirror the OpenAI
descriptionfield in the tool’sdescriptionand the function’s docstring. - Use
args_schema(a Pydantic model) for validation; it replaces the JSON Schemaparametersobject. - If your original function accepted
**kwargsor had optional fields with defaults, replicate that in the Pydantic model withField(default=...).
Verify: Import each tool in a REPL and call tool.args_schema.model_json_schema() — the output should be structurally identical to your original OpenAI parameters schema.
Step 3: replace the OpenAI call site with a LangChain chain
Wherever you currently call openai.ChatCompletion.create(functions=..., function_call=...), swap in a LangChain Runnable that binds tools to a chat model. The minimal migration uses ChatOpenAI with .bind_tools().
# services/agent.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from tools.weather import get_weather_tool
from tools.calendar import create_event_tool
from tools.search import web_search_tool
# Collect all tools in a list — order doesn't matter for the model
TOOLS = [get_weather_tool, create_event_tool, web_search_tool]
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
# If you route through a gateway like n4n.ai, set base_url and api_key here
# base_url="https://api.n4n.ai/v1",
# api_key="your-gateway-key",
).bind_tools(TOOLS)
SYSTEM_PROMPT = """You are a helpful assistant with access to tools.
Use tools when users ask for real-time data or actions."""
def run_agent(user_input: str) -> str:
messages = [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=user_input),
]
response = llm.invoke(messages)
# Handle tool calls — LangChain returns AIMessage with tool_calls attribute
if response.tool_calls:
# Execute each tool call sequentially
for tool_call in response.tool_calls:
tool = next(t for t in TOOLS if t.name == tool_call["name"])
tool_result = tool.invoke(tool_call["args"])
messages.append(response)
messages.append(tool_result) # ToolMessage with tool_call_id
# Final response after tool execution
response = llm.invoke(messages)
return response.contentIf your existing code handles parallel tool calls (OpenAI’s function_call: "auto" can return multiple), the loop above already supports it — response.tool_calls is a list.
Verify: Call run_agent("What's the weather in Tokyo?") and confirm:
- The model emits a
tool_callsblock withname: "get_weather"and correct arguments. - Your
get_weatherfunction executes and returns data. - The final response incorporates the tool result naturally.
Step 4: add structured output parsing for non-tool responses
Raw OpenAI function calling often uses a special function like respond_to_user to force structured final answers. LangChain handles this more cleanly with with_structured_output on a separate model instance, or by using a dedicated output parser. Pick one approach and apply it consistently.
# schemas.py
from pydantic import BaseModel, Field
from typing import Optional
class AgentResponse(BaseModel):
answer: str = Field(..., description="Final answer to the user")
confidence: float = Field(..., ge=0.0, le=1.0, description="Model confidence in the answer")
follow_up_needed: bool = Field(False, description="Whether the user should be prompted for more info")
citations: list[str] = Field(default_factory=list, description="Source URLs or doc IDs referenced")
# In agent.py, add a second model for structured final output
from langchain_openai import ChatOpenAI
structured_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).with_structured_output(AgentResponse)
def run_agent(user_input: str) -> AgentResponse:
messages = [
SystemMessage(content=SYSTEM_PROMPT),
HumanMessage(content=user_input),
]
response = llm.invoke(messages)
if response.tool_calls:
for tool_call in response.tool_calls:
tool = next(t for t in TOOLS if t.name == tool_call["name"])
tool_result = tool.invoke(tool_call["args"])
messages.append(response)
messages.append(tool_result)
response = llm.invoke(messages)
# Parse the final natural-language response into structured output
final_prompt = [
SystemMessage(content="Convert the assistant's final answer into the structured schema."),
HumanMessage(content=response.content),
]
return structured_llm.invoke(final_prompt)This keeps your tool-calling model focused on reasoning and tool selection, while a lighter model handles formatting.
Verify: The return value is now an AgentResponse instance. Access result.answer, result.confidence, etc. — no string parsing required.
Step 5: implement retry and fallback logic at the tool level
Raw OpenAI implementations often bury retry logic inside the HTTP call wrapper. LangChain tools support retry configuration declaratively via StructuredTool.from_function(..., retry_policy=...). This is also where you’d insert provider fallback if you route through a gateway that exposes multiple models.
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import StructuredTool
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
def get_weather_with_retry(location: str, unit: str = "fahrenheit", config: RunnableConfig = None) -> dict:
"""Fetch current weather for a location with automatic retry."""
@retry(
wait=wait_exponential_jitter(initial=1, max=10),
stop=stop_after_attempt(3),
reraise=True,
)
def _call():
import requests
resp = requests.get(
"https://api.weather.example.com/current",
params={"q": location, "units": "imperial" if unit == "fahrenheit" else "metric"},
timeout=5
)
resp.raise_for_status()
return resp.json()
return _call()
get_weather_tool = StructuredTool.from_function(
func=get_weather_with_retry,
name="get_weather",
description="Get current weather conditions for a specified location",
args_schema=GetWeatherArgs,
# Optional: add a retry policy at the Runnable level too
# retry_policy=RetryPolicy(max_attempts=3, backoff_factor=2.0),
)If your infrastructure uses a gateway that automatically fails over across providers (e.g., when a primary model is rate-limited), you don’t need tool-level fallback for model errors — but you still need it for external API failures like the weather service above.
Verify: Temporarily break the weather API endpoint (or mock a 500 response) and confirm the tool retries three times before surfacing an exception to the agent loop.
Step 6: add observability callbacks
One major advantage of migrating to LangChain is standardized callbacks for logging, tracing, and cost tracking. Attach a callback handler at the chain level to capture every tool invocation, token count, and latency without instrumenting each function manually.
# observability.py
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, Dict, List
import time
import json
import structlog
logger = structlog.get_logger()
class ToolTracingHandler(BaseCallbackHandler):
def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs) -> None:
tool_name = serialized.get("name", "unknown")
logger.info("tool_start", tool=tool_name, input=input_str, run_id=kwargs.get("run_id"))
kwargs["_start_time"] = time.perf_counter()
def on_tool_end(self, output: str, **kwargs) -> None:
tool_name = kwargs.get("name", "unknown")
duration_ms = (time.perf_counter() - kwargs.get("_start_time", 0)) * 1000
logger.info("tool_end", tool=tool_name, duration_ms=duration_ms, run_id=kwargs.get("run_id"))
def on_llm_end(self, response: LLMResult, **kwargs) -> None:
usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
logger.info("llm_complete",
model=response.llm_output.get("model_name") if response.llm_output else "unknown",
prompt_tokens=usage.get("prompt_tokens"),
completion_tokens=usage.get("completion_tokens"),
total_tokens=usage.get("total_tokens"),
run_id=kwargs.get("run_id"))
# Usage in agent.py
from observability import ToolTracingHandler
def run_agent(user_input: str, config: RunnableConfig = None) -> AgentResponse:
if config is None:
config = {}
config.setdefault("callbacks", []).append(ToolTracingHandler())
# ... rest of function unchanged, just pass config to invoke()
response = llm.invoke(messages, config=config)
# ...Verify: Check your structured logs for tool_start/tool_end pairs with matching run_id, and llm_complete entries with token counts. This replaces ad-hoc print statements and gives you a single source of truth for debugging and cost allocation.
Step 7: write integration tests that mirror your old test cases
Don’t rely on manual verification. Port your existing test fixtures — especially the ones that assert specific function call sequences — to LangChain’s testing utilities. langchain_core.runnables.utils provides assert_tool_calls helpers.
# tests/test_agent.py
import pytest
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.runnables import RunnableConfig
from services.agent import run_agent, TOOLS
from tools.weather import get_weather_tool
@pytest.fixture(autouse=True)
def mock_weather(monkeypatch):
def fake_get_weather(location: str, unit: str = "fahrenheit") -> dict:
return {"temperature": 72, "condition": "sunny", "location": location, "unit": unit}
monkeypatch.setattr(get_weather_tool, "func", fake_get_weather)
def test_weather_query_triggers_tool():
result = run_agent("What's the weather in Denver?")
# The agent should have called get_weather and returned a structured response
assert "denver" in result.answer.lower()
assert result.confidence > 0.8
assert not result.follow_up_needed
def test_parallel_tool_calls():
# If your system prompt encourages parallel calls, verify both fire
result = run_agent("Weather in Boston and Seattle?")
# This test depends on your prompt — adjust expectations accordingly
assert "boston" in result.answer.lower() or "seattle" in result.answer.lower()
def test_tool_error_handling(monkeypatch):
def failing_weather(location: str, unit: str = "fahrenheit") -> dict:
raise ConnectionError("weather API down")
monkeypatch.setattr(get_weather_tool, "func", failing_weather)
result = run_agent("Weather in Chicago?")
# Should gracefully handle the error and tell the user
assert "unavailable" in result.answer.lower() or "error" in result.answer.lower()
assert result.confidence < 0.5Run these with pytest -v. They should pass without modifying your production code.
Verify: All existing test scenarios pass. Add one new test for each tool you migrated.
Step 8: deprecate the raw OpenAI function calling code
Once tests pass and you’ve validated in staging, remove the old implementation. Delete:
- The
functions=andfunction_call=parameters from any remaining OpenAI SDK calls - Any helper modules that built JSON schemas manually
- Custom retry/wrapper code that LangChain now handles
Search your codebase for openai.ChatCompletion.create and openai.AsyncOpenAI — any call site still passing functions or tools (the new SDK parameter) is a migration candidate. If you have a wrapper class like OpenAIClient, update it to delegate to the LangChain chain instead.
Verify: grep -r "functions=" --include="*.py" . returns zero results. The test suite still passes.
Common pitfalls and how to avoid them
Schema drift: OpenAI’s JSON Schema allows additionalProperties: false by default; Pydantic v2 allows extra fields unless you set model_config = ConfigDict(extra="forbid"). Add this to every args_schema model to match the strict validation the model expects.
from pydantic import ConfigDict
class GetWeatherArgs(BaseModel):
model_config = ConfigDict(extra="forbid")
location: str = Field(...)
unit: Literal["celsius", "fahrenheit"] = Field("fahrenheit")Tool name collisions: If two tools share the same name, the model will pick arbitrarily. Enforce uniqueness in your tool registry — a simple assert len(set(t.name for t in TOOLS)) == len(TOOLS) at import time catches this.
Streaming responses: If your old code streamed tokens to the UI, use llm.astream(messages) and handle AIMessageChunk objects. Tool calls appear as chunks with tool_call_chunks — accumulate them until you have complete calls, then invoke tools, then resume streaming the final answer.
Async support: StructuredTool.from_function works with async def functions. If your original functions were async, keep them async — just add coroutine=... to the factory call or use StructuredTool.from_function(func=async_func, coroutine=async_func).
Migration checklist
- Inventory complete — every OpenAI function accounted for
- Each function converted to
StructuredToolwith matching schema - Agent chain invokes tools and handles multi-step loops
- Structured output parsing replaces ad-hoc response formatting
- Retry/fallback policies attached at tool level
- Callback handler captures tool invocations and token usage
- Integration tests ported and passing
- Legacy OpenAI function calling code removed
- Staging deployment validated with production-like traffic
The migration typically takes a few hours for a codebase with 5–10 functions. The payoff is immediate: cleaner separation of concerns, portable tool definitions that work across any LangChain-compatible model, and built-in observability you’d otherwise build yourself.