"""
LangChain OpenAI-Style API Client
==================================
A generic script for interacting with OpenAI-style chat completion APIs using LangChain.
Follows LangChain's recommended patterns and best practices.
Dependencies:
langchain>=0.1.16
langchain-openai>=0.1.0
langchain-core>=0.1.40
"""
import os
import json
from typing import Dict, List, Any, Optional
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
# ============================================================================
# Tool Definitions
# ============================================================================
@tool
def get_weather(location: str) -> str:
"""
Get weather information for a specific location.
Args:
location: The name of the location to get weather for
Returns:
A JSON string containing weather information
"""
# Mock implementation - replace with actual API call in production
weather_data = {
"temperature": 72,
"condition": "sunny",
"humidity": 65,
"location": location
}
return json.dumps(weather_data)
# ============================================================================
# Main Client Implementation
# ============================================================================
def create_chat_client(
base_url: str,
model: str,
api_key: str = "not-needed"
) -> ChatOpenAI:
"""
Create a ChatOpenAI client configured for an OpenAI-style API endpoint.
Args:
base_url: The base URL of the API endpoint (e.g., "http://localhost:9000/v1")
model: The model name to use (e.g., "qwen3_4b_instruct_2507")
api_key: API key (default: "not-needed" for local endpoints)
temperature: Temperature setting for response generation
Returns:
Configured ChatOpenAI instance
"""
# Set API key in environment (required by langchain even if not used)
os.environ["OPENAI_API_KEY"] = api_key
chat = ChatOpenAI(
base_url=base_url,
model=model
)
return chat
def process_tool_calls(
ai_message: AIMessage,
available_tools: Dict[str, Any]
) -> List[ToolMessage]:
"""
Process tool calls from the AI message and execute the corresponding tools.
Args:
ai_message: The AI message containing tool calls
available_tools: Dictionary mapping tool names to tool functions
Returns:
List of ToolMessage objects with tool execution results
"""
tool_messages = []
if hasattr(ai_message, 'tool_calls') and ai_message.tool_calls:
print("\n[TOOL CALLS DETECTED]")
for tool_call in ai_message.tool_calls:
tool_name = tool_call['name']
tool_args = tool_call['args']
tool_id = tool_call['id']
print(f" - Tool: {tool_name}")
print(f" Arguments: {json.dumps(tool_args, indent=6)}")
# Execute the tool
if tool_name in available_tools:
tool_func = available_tools[tool_name]
try:
tool_output = tool_func.invoke(tool_args)
print(f" Output: {tool_output}")
tool_messages.append(
ToolMessage(
tool_call_id=tool_id,
content=tool_output,
)
)
except Exception as e:
error_msg = f"Error executing tool {tool_name}: {str(e)}"
print(f" Error: {error_msg}")
tool_messages.append(
ToolMessage(
tool_call_id=tool_id,
content=error_msg,
)
)
else:
error_msg = f"Tool {tool_name} not found"
print(f" Error: {error_msg}")
tool_messages.append(
ToolMessage(
tool_call_id=tool_id,
content=error_msg,
)
)
return tool_messages
def run_conversation_with_tools(
chat_client: ChatOpenAI,
user_query: str,
tools: List[Any],
stream: bool = True
) -> str:
"""
Run a complete conversation with tool calling support.
Args:
chat_client: The ChatOpenAI client instance
user_query: The user's query/message
tools: List of tool functions to make available
stream: Whether to stream the response (default: True)
Returns:
The final response from the model
"""
# Bind tools to the chat client
chat_with_tools = chat_client.bind_tools(tools)
# Create tool lookup dictionary
available_tools = {tool.name: tool for tool in tools}
print("=" * 80)
print("USER QUERY")
print("=" * 80)
print(user_query)
print()
# Step 1: Send initial query
print("=" * 80)
print("MODEL RESPONSE (Initial)")
print("=" * 80)
if stream:
# Stream the response and collect chunks
chunks = []
for chunk in chat_with_tools.stream(user_query):
if chunk.content:
print(chunk.content, end="", flush=True)
chunks.append(chunk)
print()
# Reconstruct full message from chunks
if chunks:
ai_message = chunks[0]
for chunk in chunks[1:]:
ai_message += chunk
else:
print("No response received.")
return ""
else:
# Non-streaming response
ai_message = chat_with_tools.invoke(user_query)
if ai_message.content:
print(ai_message.content)
# Step 2: Check for tool calls
tool_messages = process_tool_calls(ai_message, available_tools)
# Step 3: If there were tool calls, send results back to model
if tool_messages:
print("\n" + "=" * 80)
print("SENDING TOOL RESULTS BACK TO MODEL")
print("=" * 80)
# Build complete message history
messages = [
HumanMessage(content=user_query),
ai_message,
*tool_messages
]
# Print the message history being sent
print("\nMessage History:")
for i, msg in enumerate(messages):
print(f"\n [{i}] {type(msg).__name__}")
if hasattr(msg, 'content') and msg.content:
content_preview = msg.content[:100] + "..." if len(msg.content) > 100 else msg.content
print(f" Content: {content_preview}")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f" Tool Calls: {len(msg.tool_calls)} call(s)")
if hasattr(msg, 'tool_call_id'):
print(f" Tool Call ID: {msg.tool_call_id}")
print("\n" + "=" * 80)
print("MODEL RESPONSE (Final)")
print("=" * 80)
# Get final response with tool results
if stream:
final_response = ""
for chunk in chat_with_tools.stream(messages):
if chunk.content:
print(chunk.content, end="", flush=True)
final_response += chunk.content
print()
return final_response
else:
final_message = chat_with_tools.invoke(messages)
if final_message.content:
print(final_message.content)
return final_message.content
else:
# No tool calls, return the initial response
return ai_message.content if ai_message.content else ""
return ""
# ============================================================================
# Example Usage
# ============================================================================
def main():
"""
Main function demonstrating the usage of the langchain client.
"""
# Configuration
BASE_URL = "http://localhost:9001/v1"
MODEL = "qwen3_4b_instruct_2507"
API_KEY = "not-needed" # Not required for local endpoints
print("=" * 80)
print("LANGCHAIN OPENAI-STYLE API CLIENT")
print("=" * 80)
print(f"Endpoint: {BASE_URL}")
print(f"Model: {MODEL}")
print("=" * 80)
print()
try:
# Create the chat client
chat = create_chat_client(
base_url=BASE_URL,
model=MODEL,
api_key=API_KEY
)
# Define available tools
tools = [get_weather]
# Example 1: Query that should trigger tool calling
print("\n" + "=" * 80)
print("EXAMPLE 1: Weather Query (Tool Calling Expected)")
print("=" * 80)
weather_query = "Tell me about weather in San Diego"
run_conversation_with_tools(
chat_client=chat,
user_query=weather_query,
tools=tools,
stream=False
)
except Exception as e:
print(f"\n[ERROR] An error occurred: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()
