In [1]:
from dotenv import load_dotenv
load_dotenv()
Out[1]:
True
In [2]:
import os, asyncio, json
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent, AgentState
from langchain.messages import HumanMessage, AIMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_community.utilities import SQLDatabase
from langchain_community.document_loaders import PyPDFLoader
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from langgraph.runtime import Runtime
from mcp.shared.exceptions import McpError
from mcp.types import CallToolResult, TextContent
from typing import Dict, Any
from tavily import TavilyClient
from pprint import pprint
from dataclasses import dataclass
from datetime import date
DEEPSEEK_MODEL = "deepseek-chat"
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
def deepseek_model(model: str = DEEPSEEK_MODEL, max_tokens=1000, **kwargs):
return init_chat_model(
model=model,
# DeepSeek uses LangChain's OpenAI-compatible transport.
model_provider="openai",
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=DEEPSEEK_BASE_URL,
max_tokens=max_tokens,
**kwargs,
)
Summarize messages¶
In [ ]:
from langchain.agents.middleware import SummarizationMiddleware
summarize_middleware = SummarizationMiddleware(
model=deepseek_model(),
trigger=("tokens", 100),
keep=("messages", 1)
)
agent = create_agent(
model=deepseek_model(),
checkpointer=InMemorySaver(),
middleware=[summarize_middleware]
)
In [ ]:
messages = [
HumanMessage(content="What is the capital of the moon?"),
AIMessage(content="The capital of the moon is Lunapolis."),
HumanMessage(content="What is the weather in Lunapolis?"),
AIMessage(content="Skies are clear, with a high of 120C and a low of -100C."),
HumanMessage(content="How many cheese miners live in Lunapolis?"),
AIMessage(content="There are 100,000 cheese miners living in Lunapolis."),
HumanMessage(content="Do you think the cheese miners' union will strike?"),
AIMessage(content="Yes, because they are unhappy with the new president."),
HumanMessage(content="If you were Lunapolis' new president how would you respond to the cheese miners' union?"),
]
In [ ]:
response = agent.invoke(
{"messages": messages},
{"configurable": {"thread_id": "1"}}
)
Trim/delete messages¶
In [7]:
from langchain.messages import RemoveMessage
from langchain.agents.middleware import before_agent, after_agent
@before_agent
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
"""Remove all the tool messages from the state"""
messages = state["messages"]
tool_messages = [m for m in messages if isinstance(m, ToolMessage)]
# RemoveMessage(id="abc") means remove the message with the id of abc.
return {"messages": [RemoveMessage(id=m.id) for m in tool_messages]}
In [8]:
agent = create_agent(
model=deepseek_model(),
checkpointer=InMemorySaver(),
middleware=[trim_messages],
)
In [9]:
messages = [
HumanMessage(content="My device won't turn on. What should I do?"),
ToolMessage(content="blorp-x7 initiating diagnostic ping…", tool_call_id="1"),
AIMessage(content="Is the device plugged in and turned on?"),
HumanMessage(content="Yes, it's plugged in and turned on."),
ToolMessage(content="temp=42C voltage=2.9v … greeble complete.", tool_call_id="2"),
AIMessage(content="Is the device showing any lights or indicators?"),
HumanMessage(content="What's the temperature of the device?")
]
In [10]:
response = agent.invoke(
{"messages": messages},
{"configurable": {"thread_id": "2"}}
)
pprint(response)
{'messages': [HumanMessage(content="My device won't turn on. What should I do?", additional_kwargs={}, response_metadata={}, id='577007ef-3081-41e2-8d17-eebfcb038be2'),
AIMessage(content='Is the device plugged in and turned on?', additional_kwargs={}, response_metadata={}, id='61e63e8e-d64d-4fd9-b723-4695d89d3a6d', tool_calls=[], invalid_tool_calls=[]),
HumanMessage(content="Yes, it's plugged in and turned on.", additional_kwargs={}, response_metadata={}, id='eca3565e-bc67-4154-9afb-41c632c1ab8f'),
AIMessage(content='Is the device showing any lights or indicators?', additional_kwargs={}, response_metadata={}, id='adec3dea-6170-4ca7-b926-0afbc2cf3ef9', tool_calls=[], invalid_tool_calls=[]),
HumanMessage(content="What's the temperature of the device?", additional_kwargs={}, response_metadata={}, id='f324a949-e8b5-4fe5-b673-8e093fc8b3ac'),
AIMessage(content="That's a very relevant question. The temperature of a device can tell us a lot about whether there's a serious internal issue (like a short circuit) or a less serious one (like a dead battery).\n\nPlease check the device's temperature by touching it:\n\n- **On the back** (if it's a phone/tablet)\n- **On the top or sides** (if it's a laptop or desktop computer)\n- **Near the power port** (common for overheating)\n\nWhat do you feel?\n\n- **Ice cold** (much colder than the room)\n- **Room temperature** (feels normal, like nothing is happening)\n- **Warm but not hot** (like it's been running)\n- **Hot to the touch** (uncomfortably hot, possibly burning)\n\n*Note: If the device is extremely hot, do not continue holding it. Unplug it immediately and let it cool down for 30 minutes before trying again.*", additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 194, 'prompt_tokens': 60, 'total_tokens': 254, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 60}, 'model_provider': 'openai', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_8b330d02d0_prod0820_fp8_kvcache_20260402', 'id': '950706fd-1e73-486d-b4e9-f94f2954a0fd', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019e2903-64e5-7343-9d2e-fac76313a4fc-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 60, 'output_tokens': 194, 'total_tokens': 254, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}
In [11]:
print(response["messages"][-1].content)
That's a very relevant question. The temperature of a device can tell us a lot about whether there's a serious internal issue (like a short circuit) or a less serious one (like a dead battery). Please check the device's temperature by touching it: - **On the back** (if it's a phone/tablet) - **On the top or sides** (if it's a laptop or desktop computer) - **Near the power port** (common for overheating) What do you feel? - **Ice cold** (much colder than the room) - **Room temperature** (feels normal, like nothing is happening) - **Warm but not hot** (like it's been running) - **Hot to the touch** (uncomfortably hot, possibly burning) *Note: If the device is extremely hot, do not continue holding it. Unplug it immediately and let it cool down for 30 minutes before trying again.*
In [ ]: