5 LangChain Message Types: Human, AI, System, Tool Messages

[
  {"role": "system", "content": "You are a helpful assistant."},
  {"role": "user", "content": "What's the capital of France?"},
  {"role": "assistant", "content": "The capital of France is Paris."},
  {"role": "user", "content": "What's its population?"}
]

from langchain_core.messages import (
    SystemMessage,
    HumanMessage,
    AIMessage,
    FunctionMessage,  # legacy
    ToolMessage       # modern standard
)

from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# Basic system message
system = SystemMessage(content="You are a helpful customer support agent for TechCorp. You only answer questions about TechCorp products. If asked about anything else, politely redirect to TechCorp topics.")

# System messages with structured instructions
detailed_system = SystemMessage(content="""You are an expert Python programming assistant.

Guidelines:
- Always include type hints in code examples
- Prefer f-strings over .format() or % formatting
- Suggest error handling in all database operations
- Keep explanations concise — show code first, explain second

If the user asks for code that could have security implications, explain the risk before providing it.""")

# Using system message in a conversation
from langchain_core.prompts import ChatPromptTemplate

messages = [
    system,
    HumanMessage(content="How do I connect to a PostgreSQL database?")
]

response = llm.invoke(messages)
print(response.content)

from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# Basic human message
human = HumanMessage(content="What are the benefits of serverless architecture?")

# Human message with additional_kwargs (rarely needed directly)
human_with_meta = HumanMessage(
    content="Explain quantum computing",
    additional_kwargs={"user_id": "u_12345"}  # metadata, not sent to LLM
)

# Multi-modal human message (image + text)
import base64

def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.b64encode(f.read()).decode()

multimodal_human = HumanMessage(
    content=[
        {
            "type": "text",
            "text": "What's wrong with this Python code?"
        },
        {
            "type": "image_url",
            "image_url": {
                "url": f"data:image/png;base64,{encode_image('screenshot.png')}",
                "detail": "high"
            }
        }
    ]
)

# Use in conversation
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke([SystemMessage(content="You are a code reviewer."), multimodal_human])

from langchain_core.messages import AIMessage

# Basic AI message (assistant's previous response)
ai_response = AIMessage(content="Serverless architecture eliminates server management overhead, scales automatically with demand, and follows a pay-per-execution pricing model.")

# AI message with tool calls (generated by the model)
ai_with_tool_call = AIMessage(
    content="",  # content is empty when the model calls a tool
    tool_calls=[
        {
            "id": "call_abc123",
            "name": "search_web",
            "args": {"query": "LangChain latest release 2026"}
        }
    ]
)

# Reconstructing conversation history
conversation = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="What is the capital of France?"),
    AIMessage(content="The capital of France is Paris."),
    HumanMessage(content="What's its population?"),
    # The LLM uses all previous messages to answer this follow-up
]

llm = ChatOpenAI(model="gpt-4o")
follow_up_response = llm.invoke(conversation)
print(follow_up_response.content)
# → Uses "France" and "Paris" from context to answer the population question

from langchain_core.messages import ToolMessage, AIMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# Define a tool
@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    # In production, call a real weather API
    return f"The weather in {city} is 22°C, partly cloudy."

@tool
def get_population(city: str) -> str:
    """Get population of a city."""
    populations = {"Paris": "2.16 million", "London": "8.9 million"}
    return populations.get(city, "Population data not available")

# Bind tools to LLM
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools([get_weather, get_population])

# Conversation with tool use
messages = [
    SystemMessage(content="You are a helpful assistant with access to weather and population data."),
    HumanMessage(content="What's the weather and population of Paris?")
]

# Step 1: LLM decides to call tools
response = llm_with_tools.invoke(messages)
print(f"Tool calls: {response.tool_calls}")
# → [{'name': 'get_weather', 'args': {'city': 'Paris'}, 'id': 'call_xyz'}]

messages.append(response)  # Add AIMessage with tool calls to history

# Step 2: Execute tools and add ToolMessages
for tool_call in response.tool_calls:
    if tool_call["name"] == "get_weather":
        result = get_weather.invoke(tool_call["args"])
    elif tool_call["name"] == "get_population":
        result = get_population.invoke(tool_call["args"])
    else:
        result = "Tool not found"
    
    messages.append(ToolMessage(
        content=result,
        tool_call_id=tool_call["id"]  # must match the AIMessage tool_call id
    ))

# Step 3: LLM synthesizes tool results into a final answer
final_response = llm_with_tools.invoke(messages)
print(final_response.content)
# → "Paris currently has a temperature of 22°C with partly cloudy skies,
#    and a population of approximately 2.16 million people."

from langchain_core.messages import FunctionMessage
import json

# Legacy pattern — avoid in new code
legacy_function_result = FunctionMessage(
    name="get_weather",
    content=json.dumps({"temperature": 22, "condition": "partly cloudy"})
)

# Modern equivalent
modern_tool_result = ToolMessage(
    content=json.dumps({"temperature": 22, "condition": "partly cloudy"}),
    tool_call_id="call_abc123"
)

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from typing import List, Union
import json

MessageType = Union[SystemMessage, HumanMessage, AIMessage, ToolMessage]

@tool
def search_database(query: str) -> str:
    """Search the product database for relevant items."""
    # Mock database
    products = {
        "laptop": "ThinkPad X1 Carbon - $1,299 - In stock",
        "keyboard": "Mechanical KB Pro - $149 - In stock",
        "monitor": "4K Display 27\" - $599 - Limited stock"
    }
    for key, value in products.items():
        if key in query.lower():
            return value
    return "No matching products found."

@tool
def check_availability(product_id: str) -> str:
    """Check real-time availability for a product."""
    return f"Product {product_id}: 47 units available, ships in 2-3 business days."

class ConversationManager:
    def __init__(self, system_prompt: str):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = [search_database, check_availability]
        self.llm_with_tools = self.llm.bind_tools(self.tools)
        self.messages: List[MessageType] = [
            SystemMessage(content=system_prompt)
        ]

    def process_tool_calls(self, ai_message: AIMessage) -> None:
        """Execute tool calls and append results to message history."""
        tool_map = {t.name: t for t in self.tools}

        for tool_call in ai_message.tool_calls:
            tool_name = tool_call["name"]
            tool_args = tool_call["args"]
            tool_id = tool_call["id"]

            if tool_name in tool_map:
                result = tool_map[tool_name].invoke(tool_args)
            else:
                result = f"Unknown tool: {tool_name}"

            self.messages.append(ToolMessage(
                content=str(result),
                tool_call_id=tool_id
            ))

    def chat(self, user_input: str) -> str:
        # Add user message
        self.messages.append(HumanMessage(content=user_input))

        # Agentic loop: keep running until no more tool calls
        while True:
            response = self.llm_with_tools.invoke(self.messages)
            self.messages.append(response)

            if response.tool_calls:
                self.process_tool_calls(response)
                # Continue loop — let LLM process tool results
            else:
                return response.content

# Usage
agent = ConversationManager(
    system_prompt="You are a helpful e-commerce assistant. Use tools to look up products and check availability."
)

print(agent.chat("I'm looking for a laptop"))
print(agent.chat("Is it in stock? Can you check?"))
print(agent.chat("What's the keyboard you have?"))

from langchain_core.messages import trim_messages, SystemMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

# Build a long conversation
messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="Tell me about quantum computing"),
    AIMessage(content="Quantum computing uses quantum mechanical phenomena..."),
    HumanMessage(content="How does entanglement work?"),
    AIMessage(content="Quantum entanglement is a phenomenon where..."),
    HumanMessage(content="What are the practical applications?"),
    AIMessage(content="Current practical applications include..."),
    HumanMessage(content="What companies are working on this?"),
    AIMessage(content="Major players include IBM, Google, Microsoft..."),
    HumanMessage(content="What's the latest from IBM?")
]

# Trim to fit within token budget
trimmed = trim_messages(
    messages,
    max_tokens=2000,
    strategy="last",           # keep most recent messages
    token_counter=llm,         # use the model's tokenizer
    include_system=True,       # always keep the system message
    allow_partial=False,       # don't cut messages mid-way
    start_on="human"           # ensure first non-system message is human
)

print(f"Original: {len(messages)} messages")
print(f"Trimmed: {len(trimmed)} messages")

# The trimmed list always starts with SystemMessage + a HumanMessage
for msg in trimmed:
    print(f"{type(msg).__name__}: {msg.content[:80]}...")

from langchain_core.messages import filter_messages, AIMessage, HumanMessage

messages = [
    SystemMessage(content="System instruction"),
    HumanMessage(content="First question"),
    AIMessage(content="First answer"),
    HumanMessage(content="Second question"),
    AIMessage(content="Second answer"),
]

# Get only human messages
human_only = filter_messages(messages, include_types=[HumanMessage])
print(f"Human messages: {len(human_only)}")

# Exclude system messages
no_system = filter_messages(messages, exclude_types=[SystemMessage])
print(f"Non-system messages: {len(no_system)}")

# Get by role string
tool_messages = filter_messages(messages, include_types=["tool"])

from langchain_core.messages import messages_to_dict, messages_from_dict
import json

messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="Hello!"),
    AIMessage(content="Hi there! How can I help?"),
]

# Serialize to JSON-compatible dict
serialized = messages_to_dict(messages)
json_str = json.dumps(serialized)
print(json_str[:200])

# Store to database (example with a simple file store)
def save_conversation(session_id: str, messages):
    serialized = messages_to_dict(messages)
    with open(f"conversations/{session_id}.json", "w") as f:
        json.dump(serialized, f)

def load_conversation(session_id: str):
    with open(f"conversations/{session_id}.json", "r") as f:
        serialized = json.load(f)
    return messages_from_dict(serialized)

# Round-trip test
save_conversation("session_001", messages)
restored = load_conversation("session_001")

assert len(restored) == len(messages)
assert type(restored[0]) == SystemMessage
assert type(restored[1]) == HumanMessage
print("Serialization round-trip successful")

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Template that preserves conversation history
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a {persona}. Your expertise is {domain}."),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

# Render the template
filled = chat_prompt.invoke({
    "persona": "senior software engineer",
    "domain": "distributed systems",
    "chat_history": [
        HumanMessage(content="What's CAP theorem?"),
        AIMessage(content="CAP theorem states that a distributed system can only guarantee two of three properties...")
    ],
    "input": "Which databases sacrifice consistency?"
})

print(type(filled.messages[0]))  # SystemMessage
print(type(filled.messages[1]))  # HumanMessage (from history)
print(type(filled.messages[2]))  # AIMessage (from history)
print(type(filled.messages[3]))  # HumanMessage (current input)

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def lookup_order(order_id: str) -> str:
    """Look up order status by order ID."""
    return f"Order {order_id}: Shipped on May 28, estimated delivery June 2."

@tool
def create_refund_request(order_id: str, reason: str) -> str:
    """Create a refund request for an order."""
    return f"Refund request created for order {order_id}. Ticket #REF-{order_id[-4:]}. Processing 3-5 business days."

SUPPORT_SYSTEM = SystemMessage(content="""You are a customer support agent for TechStore.

Rules:
- Always ask for the order ID before looking up orders
- Be empathetic and professional
- Offer refunds for orders damaged in transit
- Never make up information — use tools to get accurate data
- End responses with "Is there anything else I can help you with?"
""")

class SupportAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
        self.tools = [lookup_order, create_refund_request]
        self.llm_with_tools = self.llm.bind_tools(self.tools)
        self.history = [SUPPORT_SYSTEM]

    def respond(self, customer_input: str) -> str:
        self.history.append(HumanMessage(content=customer_input))

        while True:
            response = self.llm_with_tools.invoke(self.history)
            self.history.append(response)

            if not response.tool_calls:
                return response.content

            tool_map = {t.name: t for t in self.tools}
            for tc in response.tool_calls:
                result = tool_map[tc["name"]].invoke(tc["args"])
                self.history.append(ToolMessage(
                    content=result,
                    tool_call_id=tc["id"]
                ))

agent = SupportAgent()
print(agent.respond("My order arrived damaged, order #TS-98765"))
print(agent.respond("Yes please create the refund"))