Agent Development

from dotenv import load_dotenv

load_dotenv()

def get_default_model():
    if os.getenv("OPENAI_API_KEY"):
        return "gpt-4-turbo"
    elif os.getenv("GOOGLE_API_KEY"):
        return "google:gemini-1.5-flash"
    elif os.getenv("ANTHROPIC_API_KEY"):
        return "anthropic:claude-sonnet-4-5"
    elif os.getenv("GROQ_API_KEY"):
        return "groq:llama-3.3-70b-versatile"
    else:
        raise ValueError("No API key found!")

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

# messages = [
#     {"role": "system", "content": "you are a helpful assistant."},
#     {"role": "user", "content": "Hello! How are you?"},
# ]
messages = [
    SystemMessage(content="you are a helpful assistant."),
    HumanMessage(content="Hello! How are you?"),
]
# print("messages:", messages)
print(messages[-1].type, ": ", messages[-1].content)

# sync
try:
    response = model.invoke(messages, config=None)  # can be string as well
    messages.append(response)
    print(messages[-1].type, ": ", messages[-1].content)
except ValueError as e:
    print("configuration error:", e)
except ConnectionError as e:
    print("network error:", e)
except Exception as e:
    print("unknown error:", type(e).__name__, ":", e)

# streaming
# for chunk in model.stream(messages):
#     print(chunk.content, end="", flush=True)

{
    "content": "\nHello! 😊 I'm doing great, thanks for asking! How are you doing today? I'm here to help with anything you need!",
    "additional_kwargs": {
        "refusal": null
    },
    "response_metadata": {
        "token_usage": {
            "completion_tokens": 197,
            "prompt_tokens": 15,
            "total_tokens": 212,
            "completion_tokens_details": {
                "accepted_prediction_tokens": null,
                "audio_tokens": null,
                "reasoning_tokens": 167,
                "rejected_prediction_tokens": null
            },
            "prompt_tokens_details": null
        },
        "model_provider": "openai",
        "model_name": "deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
        "system_fingerprint": "",
        "id": "019b53c7558eafd3ce017b75c61a3841",
        "finish_reason": "stop",
        "logprobs": null
    },
    "type": "ai",
    "name": null,
    "id": "lc_run--019b53c7-53ef-7172-a335-aa785cad47ba-0",
    "tool_calls": [],
    "invalid_tool_calls": [],
    "usage_metadata": {
        "input_tokens": 15,
        "output_tokens": 197,
        "total_tokens": 212,
        "input_token_details": {},
        "output_token_details": {
            "reasoning": 167
        }
    }
}

import os
import base64
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage


def encode_image_to_base64(image_path: str) -> str:
    with open(image_path, "rb") as image_file:
        encoded_string = base64.b64encode(image_file.read()).decode("utf-8")
    return encoded_string


def get_mime_type(image_path: str) -> str:
    ext = os.path.splitext(image_path)[1].lower()
    if ext in [".jpg", ".jpeg"]:
        return "image/jpeg"
    elif ext == ".png":
        return "image/png"
    elif ext == ".gif":
        return "image/gif"
    elif ext == ".webp":
        return "image/webp"
    else:
        return "application/octet-stream"


load_dotenv()

# support multimodal input
model = init_chat_model(
    model_provider="openai",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    model="qwen-vl-max",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

img_path = "./test.png"
messages = [
    SystemMessage(content="you are a helpful assistant."),
    HumanMessage(
        content=[
            {"type": "text", "text": "describe the following image simply."},
            # {
            #     "type": "image_url",
            #     "image_url": {"url": "https://example.com/path/to/your/image.jpg"},
            # },
            {
                "type": "image_url",
                "image_url": {
                    "url": f"data:{get_mime_type(img_path)};base64,{encode_image_to_base64(img_path)}"
                },
            },
            # ...
        ]
    ),
]

response = model.invoke(messages, config=None)
messages.append(response)
print(messages[-1].type, ": ", messages[-1].content)

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain_core.prompts import (
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
    AIMessagePromptTemplate,
)

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

# prompt
template = PromptTemplate.from_template(
    "You are a helpful assistant that translates {input_language} to {output_language}."
)
prompt_demo1 = template.format(input_language="English", output_language="Chinese")
prompt_demo2 = template.invoke(
    {"input_language": "English", "output_language": "Chinese"}
)
# messages = prompt_demo2.to_messages()
print("Template string: ", template.template)
print("Template variables: ", template.input_variables)
print("Template format: ", prompt_demo1)
print("Template format: ", prompt_demo2.text)


# messages
# chat_template = ChatPromptTemplate.from_messages(
#     [
#         (
#             "system",
#             "You are a helpful assistant that translates {input_language} to {output_language}.",
#         ),
#         ("user", "Translate the following text: {text}"),
#     ]
# )
system_message = SystemMessagePromptTemplate.from_template(
    "You are a helpful assistant that translates {input_language} to {output_language}."
)
human_message = HumanMessagePromptTemplate.from_template(
    "Translate the following text: {text}"
)
chat_template = ChatPromptTemplate.from_messages(
    [
        system_message,
        human_message,
    ]
)
# chat_template = ChatPromptTemplate.from_messages([system_message + human_message])

# partical_chat_template = chat_template.partial(
#     input_language="English", output_language="Chinese"
# )
# messages = partical_chat_template.format_messages(
#     text="How are you today?",
# )
messages = chat_template.format_messages(
    input_language="English",
    output_language="Chinese",
    text="How are you today?",
)
for msg in messages:
    print(f"{msg.type}: {msg.content}")

response = model.invoke(messages, config=None)
messages.append(response)
print(messages[-1].type, ": ", messages[-1].content)

from langchain_core.prompts import ChatPromptTemplate


class PromptLibrary:

    TRANSLATOR = ChatPromptTemplate.from_template(
        (
            "system",
            "You are a helpful assistant that translates {input_language} to {output_language}.",
        ),
        ("user", "Translate the following text: {text}"),
    )

    SUMMARIZER = ChatPromptTemplate.from_template(
        ("system", "You are a concise summarization assistant."),
        ("user", "Summarize the following text: {text}"),
    )

    QA = ChatPromptTemplate.from_template(
        (
            "system",
            "You are an expert in answering questions based on provided context.",
        ),
        (
            "user",
            "Based on the following context, answer the question: {context}\nQuestion: {question}",
        ),
    )

    SENTIMENT_ANALYZER = ChatPromptTemplate.from_template(
        ("system", "You are a sentiment analysis assistant."),
        ("user", "Analyze the sentiment of the following text: {text}"),
    )

    CODE_EXPLAINER = ChatPromptTemplate.from_template(
        ("system", "You are an expert programmer who explains code clearly."),
        ("user", "Explain the following code: {code_snippet}"),
    )


messages = PromptLibrary.TRANSLATOR.format_messages(
    input_language="English",
    output_language="French",
    text="Hello, how are you?",
)

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
from langchain_core.prompts import (
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
    AIMessagePromptTemplate,
)

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

chat_template = ChatPromptTemplate.from_messages(
    [
        SystemMessagePromptTemplate.from_template(
            "You are a helpful assistant that translates {input_language} to {output_language}."
        ),
        HumanMessagePromptTemplate.from_template(
            "Translate the following text: {text}"
        ),
    ]
)

chain = chat_template | model
response = chain.invoke(
    {
        "input_language": "English",
        "output_language": "Chinese",
        "text": "Hello, how are you?",
    }
)
print(response.content)

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.prompts import (
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
    AIMessagePromptTemplate,
)

load_dotenv()


def clip_recent_messages(messages, max_pairs=5):
    if len(messages) <= max_pairs * 2 + 1:
        return messages

    system_msgs = [msg for msg in messages if msg.type == "system"]
    conversation_msgs = [msg for msg in messages if msg.type != "system"]

    recent_conversation_msgs = conversation_msgs[-(max_pairs * 2) :]

    return system_msgs + recent_conversation_msgs


model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

system_message = SystemMessagePromptTemplate.from_template(
    "You are a helpful assistant that translates {input_language} to {output_language}."
)
human_message = HumanMessagePromptTemplate.from_template(
    "Translate the following text: {text}"
)
messages = system_message.format_messages(
    input_language="English",
    output_language="Chinese",
)

while True:
    input_text = input("Enter text to translate (or 'exit' to quit): ")
    if input_text.lower() == "exit":
        break

    messages = clip_recent_messages(messages, max_pairs=2)
    user_msg = human_message.format_messages(
        text=input_text,
    )
    messages.append(user_msg[0])

    response = model.invoke(messages, config=None)
    messages.append(response)
    print(messages[-1].type, ": ", messages[-1].content)

Response: {
    "messages": [
        HumanMessage(
            content="What time is it now? Don't tell me the seconds.",
        ),
        AIMessage(
            tool_calls=[
                {
                    "name": "get_current_time",
                    "args": {"without_seconds": True},
                    "id": "cnnv06vs4",
                    "type": "tool_call",
                }
            ],
        ),
        ToolMessage(
            content="2025-12-25 19:11",
            name="get_current_time",
            id="5b6931f8-de26-4c07-9ab5-16503fd9840e",
            tool_call_id="cnnv06vs4",
        ),
        AIMessage(
            content="The current time is 19:11.",
        ),
    ]
}

# 简单场景，CPU 密集型任务
import os
import time
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from pydantic import BaseModel, Field, field_validator
from langchain_core.tools import StructuredTool

load_dotenv()


# Define args schema
class GetCurrentTimeArgs(BaseModel):
    """
    获取当前的日期和时间的参数
    """

    without_seconds: bool = Field(
        default=False,
        description="是否不显示秒数，默认为 False",
    )


# 1. Define a tool to get the current time
@tool
def get_current_time(without_seconds: bool = False) -> str:
    """
    获取当前的日期和时间

    参数：
        without_seconds: 是否不显示秒数，默认为 False

    返回:
        当前的日期和时间字符串
    """
    from datetime import datetime

    try:
        if without_seconds:
            return datetime.now().strftime("%Y-%m-%d %H:%M")
        else:
            return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    except Exception as e:
        return f"get_current_time error: {str(e)}"


# 2 Define a tool to get the current time[args validation]
@tool(args_schema=GetCurrentTimeArgs)
def get_current_time_args_validation(without_seconds: bool = False) -> str:
    """
    获取当前的日期和时间

    返回:
        当前的日期和时间字符串
    """
    from datetime import datetime

    try:
        if without_seconds:
            return datetime.now().strftime("%Y-%m-%d %H:%M")
        else:
            return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    except Exception as e:
        return f"get_current_time error: {str(e)}"


# 3 Define a tool to get the current time[StructuredTool]
def get_current_time_structured(without_seconds: bool = False) -> str:
    from datetime import datetime

    try:
        if without_seconds:
            return datetime.now().strftime("%Y-%m-%d %H:%M")
        else:
            return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    except Exception as e:
        return f"get_current_time error: {str(e)}"


get_current_time_tool = StructuredTool.from_function(
    func=get_current_time_structured,
    name="get_current_time_structured",
    description="\n获取当前的日期和时间\n\n返回:\n        当前的日期和时间字符串\n",
    args_schema=GetCurrentTimeArgs,
)

# Test invoke
res = get_current_time.invoke({"without_seconds": True})
print("<get_current_time> invoke result:", res)
print("<get_current_time> name:", get_current_time.name)
print("<get_current_time> description:", get_current_time.description)
print("<get_current_time> args:", get_current_time.args)
exit(0)

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.groq.com/openai/v1",
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)
model_with_time = model.bind_tools([get_current_time])

response = model_with_time.invoke("What time is it now? Don't tell me the seconds.")
print(response.type, ": ")
if hasattr(response, "tool_calls"):
    for call in response.tool_calls:
        print(call["name"], ":", call["args"])
else:
    print(response.content)

# tool_calls=[
#     {
#         name: get_current_time,
#         args: {
#             without_seconds: True
#         },
#         id: 2e6hyt315,
#         type: tool_call
#     }
# ]


# tools factory
def create_category_search(category: str) -> StructuredTool:
    class SearchArgs(BaseModel):
        query: str = Field(..., description="搜索关键词")

        @field_validator("query")
        def validate_query(cls, v):
            if not v or len(v.strip()) == 0:
                raise ValueError("查询关键词不能为空")
            return v

    @tool
    def search_products(query: str) -> dict:
        sample_data = {
            "电子产品": [
                {"name": "iPhone 14", "price": 7999},
                {"name": "Samsung Galaxy S22", "price": 6999},
                {"name": "Sony WH-1000XM4", "price": 2499},
            ],
            "服装": [
                {"name": "Levi's 牛仔裤", "price": 499},
                {"name": "Nike 运动鞋", "price": 699},
                {"name": "Adidas 运动衫", "price": 399},
            ],
        }

        results = [
            item
            for item in sample_data.get(category, [])
            if query.lower() in item["name"].lower()
        ]

        return {"category": category, "results": results}

    return StructuredTool.from_function(
        func=search_products,
        name=f"search_{category.lower()}",
        description=f"\n搜索类别为 '{category}' 的商品\n\n参数:\n    query: 搜索关键词\n\n返回:\n    包含搜索结果的字典\n",
        args_schema=SearchArgs,
    )

# IO 密集型（API 调用、数据库、文件操作）
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
import asyncio
import aiohttp
from langchain_core.tools import tool

load_dotenv()


@tool
async def async_fetch_url(url: str) -> str:
    """
    异步获取指定URL的内容

    参数：
        url: 目标URL

    返回:
        URL的内容字符串
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()


model = init_chat_model(
    model_provider="openai",
    base_url="https://api.groq.com/openai/v1",
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

model_with_fetch = model.bind_tools([async_fetch_url])


# Test invoke
async def test_invoke():
    res = await async_fetch_url.ainvoke({"url": "https://chat.deepseek.com/"})
    print("<async_fetch_url> invoke result:", res[:10], "...")
    print("<async_fetch_url> name:", async_fetch_url.name)
    print("<async_fetch_url> description:", async_fetch_url.description)
    print("<async_fetch_url> args:", async_fetch_url.args)


async def main():
    await test_invoke()

    response = await model_with_fetch.ainvoke(
        "give me content from url: https://chat.deepseek.com/."
    )
    print(response.type, ": ")
    if hasattr(response, "tool_calls"):
        for call in response.tool_calls:
            print(call["name"], ":", call["args"])
    else:
        print(response.content)


if __name__ == "__main__":
    asyncio.run(main())

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool

load_dotenv()


@tool
def get_current_time(without_seconds: bool = False) -> str:
    """
    获取当前的日期和时间

    参数：
        without_seconds: 是否不显示秒数，默认为 False

    返回:
        当前的日期和时间字符串
    """
    # importance of docstring for tool description
    from datetime import datetime

    try:
        if without_seconds:
            return datetime.now().strftime("%Y-%m-%d %H:%M")
        else:
            return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    except Exception as e:
        return f"get_current_time error: {str(e)}"


model = init_chat_model(
    model_provider="openai",
    base_url="https://api.groq.com/openai/v1",
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

agent = create_agent(
    model=model,
    tools=[get_current_time],  # only provide the necessary tools
    system_prompt="You are an assistant who answers concisely and clearly.",
)

messages = [
    {
        "role": "user",
        "content": "What time is it now? Don't tell me the seconds.",
    }
]
response = agent.invoke(
    {"messages": messages}
)  # give messages of this epoch conversation
for msg in response["messages"]:
    print(msg.type, ": ")
    if hasattr(msg, "content") and msg.content:
        print(msg.content)
    elif hasattr(msg, "tool_calls") and msg.tool_calls:
        for call in msg.tool_calls:
            print(call["name"], ":", call["args"])
    elif hasattr(msg, "name"):
        print(msg["name"], ":", msg["content"])
    else:
        pass

print("\n--- Next Question ---\n")

messages = response["messages"] + [
    {
        "role": "user",
        "content": "next question: what is the time in New York?",
    }
]
response2 = agent.invoke({"messages": messages})
for msg in response2["messages"]:
    print(msg.type, ": ")
    if hasattr(msg, "content") and msg.content:
        print(msg.content)
    elif hasattr(msg, "tool_calls") and msg.tool_calls:
        for call in msg.tool_calls:
            print(call["name"], ":", call["args"])
    elif hasattr(msg, "name"):
        print(msg["name"], ":", msg["content"])
    else:
        pass

import os
import time
from dotenv import load_dotenv
from langchain.messages import HumanMessage, ToolMessage
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict

load_dotenv()


@tool
def get_time():
    """获取时间"""
    from datetime import datetime

    return f"{datetime.now().strftime('%H:%M')}"


@tool
def add(a: int, b: int):
    """加法计算"""
    return f"{a} + {b} = {a + b}"


class ToolMonitor(BaseCallbackHandler):
    def __init__(self):
        self._start_time = None
        self.tool_stats = {}

    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs):
        self._start_time = time.time()

        tool_name = serialized.get("name", "unknown_tool")
        print(f"{tool_name}调用: {input_str[:30]}...")

        if tool_name not in self.tool_stats:
            self.tool_stats[tool_name] = {
                "success_calls": 0,
                "error_calls": 0,
                "avg_duration": 0.0,
            }

    def on_tool_end(self, output: ToolMessage, **kwargs):
        cur_duration = time.time() - self._start_time
        tool_name = kwargs.get("name", "unknown_tool")
        prev_avg_duration = self.tool_stats[tool_name]["avg_duration"]
        print(f"{tool_name}耗时: {cur_duration:.2f} 秒完成:{output.content[:30]}...")

        self.tool_stats[tool_name]["success_calls"] += 1
        self.tool_stats[tool_name]["avg_duration"] = (
            cur_duration + prev_avg_duration
        ) / 2.0

    def on_tool_error(self, error: Exception, **kwargs):
        tool_name = kwargs.get("name", "unknown_tool")
        print(f"{tool_name}调用出错: {error}")

        self.tool_stats[tool_name]["error_calls"] += 1

    def get_stats(self):
        for tool_name, stats in self.tool_stats.items():
            print(
                f"工具: {tool_name} | 成功调用: {stats['success_calls']} | 错误调用: {stats['error_calls']} | 平均耗时: {stats['avg_duration']:.2f} 秒"
            )

    def on_chain_end(self, outputs, *, run_id, parent_run_id=None, **kwargs):
        # print(outputs)
        pass


model = init_chat_model(
    model_provider="openai",
    base_url="https://api.groq.com/openai/v1",
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)
agent = create_agent(
    model=model,
    tools=[get_time, add],
    system_prompt="你是一个有用的助手，可以使用工具来帮助用户。",
)
monitor = ToolMonitor()

questions = ["现在几点？", "计算 5+3"]
for q in questions:
    print(f"问题: {q}")
    response = agent.invoke(
        {"messages": [HumanMessage(content=q)]}, config={"callbacks": [monitor]}
    )
    print(f"回答: {response['messages'][-1].content}")
    print("-" * 20)

print("\n工具调用统计:")
monitor.get_stats()

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.agents import create_agent

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.groq.com/openai/v1",
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

agent = create_agent(
    model=model,
    system_prompt="You are an assistant who answers concisely and clearly.",
)

# streaming
for chunk in agent.stream(
    {"messages": [HumanMessage(content="Hello! How are you?")]}, stream_mode="messages"
):
    # print(chunk["model"]["messages"][-1].content)  # stream_mode="updates"

    if chunk[0].content is not None:
        print(chunk[0].content, end="", flush=True)  # stream_mode="messages"

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.groq.com/openai/v1",
    model="llama-3.3-70b-versatile",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)

agent = create_agent(
    model=model,
    system_prompt="You are an assistant who answers concisely and clearly.",
    checkpointer=InMemorySaver(),
)
# {"thread_id": "xxx", "messages": [HumanMessage("..."), AIMessage("..."), ...]}


contents = [
    "Hello! I am John.",
    "Can you tell me my name?",
]
config = {"configurable": {"thread_id": "conversation_1"}}
# config2 = {"configurable": {"thread_id": "conversation_2"}}  # different context

for content in contents:
    response = agent.invoke(
        {"messages": [HumanMessage(content=content)]}, config=config
    )  # history in response
    print(response["messages"][-2].type, ": ", response["messages"][-2].content)
    print(response["messages"][-1].type, ": ", response["messages"][-1].content)

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.agents import create_agent
from langgraph.checkpoint.sqlite import SqliteSaver

"""
1. invoke 前：LangGraph 自动调用 checkpointer.get(thread_id="user_123")      
    - 查询数据库，读取该 thread_id 的历史消息
    - 如果是第一次，返回空列表
2. invoke 中：Agent 处理时会看到完整历史(默认确实会全部读取，)
    state = {
        "messages": [历史消息1, 历史消息2, 新消息]  # 自动合并
    }
3. invoke 后：LangGraph 自动调用 checkpointer.put(thread_id, state)
    - 将新的完整状态写入数据库
    - 数据库存储：(thread_id, timestamp, messages)
"""

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)
db_path = "./db/checkpoints.sqlite"

with SqliteSaver.from_conn_string(db_path) as checkpointer:  # auto manage connect
    agent = create_agent(
        model=model,
        system_prompt="You are an assistant who answers concisely and clearly.",
        checkpointer=checkpointer,
    )
    config = {"configurable": {"thread_id": "conversation_1"}}

    contents = [
        "Hello! I am John Steve Take Z.",
        "Can you tell me my name?",
    ]
    for content in contents:
        response = agent.invoke(
            {"messages": [HumanMessage(content=content)]}, config=config
        )
        print(response["messages"][-2].type, ": ", response["messages"][-2].content)
        print(response["messages"][-1].type, ": ", response["messages"][-1].content)


# clean on time, or backup, or archive
# import sqlite3

# conn = sqlite3.connect(db_path)
# cursor = conn.cursor()
# cursor.execute("DELETE FROM checkpoints WHERE thread_id = ?", ("conversation_1",))
# conn.commit()
# conn.close()

from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.messages import trim_messages

messages = [
    HumanMessage(content="message 1"),
    AIMessage(content="response 1"),
    HumanMessage(content="message 2"),
    AIMessage(content="response 2"),
    HumanMessage(content="message 3"),
    AIMessage(content="response 3"),
]
print("Original messages length:", len(messages))

trimmed = trim_messages(
    messages,
    max_tokens=4,
    token_counter=len,
    strategy="last",
)
print("Trimmed messages length:", len(trimmed))
for msg in trimmed:
    print(f"{msg.type}: {msg.content}")

用户输入 (HumanMessage)
            |
            v
+---------------------------------------+
|          agent.invoke(input)          |
+---------------------------------------+
            |
            | 1. 获取当前状态 (State)
            v
    [  Checkpointer (内存/数据库) ] <--- 加载历史消息
            |
            v
+---------------------------------------+
|  Middleware: before_model (按顺序执行)   | <--- 修改 state
+---------------------------------------+
            |
            | 2. 准备好的 Messages 发送给 LLM
            v
+---------------------------------------+
|           LLM (大语言模型)             |
|  (接收处理后的消息，生成 AIMessage)      |
+---------------------------------------+
            |
            | 3. 得到模型回复
            v
+---------------------------------------+
|  Middleware: after_model (逆序执行)    | <--- state 已包含 AI 回复
+---------------------------------------+
            |
            | 4. 状态归约 (State Reducer)
            v
+---------------------------------------+
|    messages.add(new_messages)         | <--- 默认是“追加”而非“覆盖”
+---------------------------------------+
            |
            | 5. 保存状态到 Checkpointer
            v
+---------------------------------------+
|          返回最终 Response             |
+---------------------------------------+

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.agents import create_agent
from langchain.agents.middleware import AgentState, AgentMiddleware, hook_config
from langgraph.runtime import Runtime
from langchain_core.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import NotRequired
from typing import Any

"""
what can do before model?
- logging
- formatting
- trimming
- validation

what can return after model? [append mode]
- None: no changes
- dict: update state
- {"jump_to": "step_name"}: jump to specific step
    - "end": Jump to the end of the agent execution (or the first after_agent hook)
    - "tools":  Jump to the tools node
    - "model": Jump to the model node (or the first before_model hook)
"""


class CounterMiddleware(AgentMiddleware):
    # maintain state in middleware instance
    def __init__(self, max_calls: int = 5):
        super().__init__()
        self.count = 0
        self.max_calls = max_calls

    def after_model(self, state, runtime) -> None | dict[str, Any]:
        self.count += 1
        print("> Model call count in after_model(middleware inner):", self.count)
        if self.count >= self.max_calls:
            raise ValueError("Exceeded maximum model call counts.")
        return None


class CustomState(AgentState):
    model_call_count: NotRequired[int]
    user_id: NotRequired[str]


class CallsLimitMiddleware(AgentMiddleware[CustomState]):
    # maintain state in CustomState
    def __init__(self, max_calls: int = 5):
        super().__init__()
        self.max_calls = max_calls

    @hook_config(can_jump_to=["end"])
    def before_model(
        self, state: CustomState, runtime: Runtime
    ) -> None | dict[str, Any]:
        count = state.get("model_call_count", 0)
        if count >= self.max_calls:
            return {
                "messages": [AIMessage(content="Model call limit reached.")],
                "jump_to": "end",
            }
        return {"model_call_count": count}

    def after_model(
        self, state: CustomState, runtime: Runtime
    ) -> None | dict[str, Any]:
        count = state["model_call_count"] + 1
        print("> Model call count in after_model(custom state):", count)
        return {"model_call_count": count}


class MessagesTrimMiddleware(AgentMiddleware):
    # must without checkpointer, and manual combine messages(response & new input)
    def __init__(self, max_messages: int = 20) -> None:
        super().__init__()
        self.max_messages = max_messages

    def before_model(self, state, runtime) -> None | dict[str, Any]:
        messages = state["messages"]
        if len(messages) > self.max_messages:
            trimmed_messages = messages[-self.max_messages :]
            print(f"> Trimming messages from {len(messages)} to {self.max_messages}.")
            return {"messages": trimmed_messages}
        return None


class MessagesLimitMiddleware(AgentMiddleware):
    # with checkpointer, use RemoveMessage to remove old messages in checkpointer
    def __init__(self, max_messages: int = 20) -> None:
        super().__init__()
        self.max_messages = max_messages

    def before_model(self, state, runtime) -> None | dict[str, Any]:
        messages = state["messages"]
        if len(messages) > self.max_messages:
            to_remove = [
                RemoveMessage(id=m.id)
                for m in messages[: (len(messages) - self.max_messages)]
            ]
            if to_remove:
                print(f"> Trimming {len(to_remove)} messages from the beginning.")
                return {"messages": to_remove}

        return None


class LoggingMiddleware(AgentMiddleware):
    def after_model(self, state, runtime) -> None | dict[str, Any]:
        messages = state["messages"]
        print("----------------------------")
        for msg in messages:
            print(msg.type, ": ", msg.content)
        print("----------------------------")

        return None


load_dotenv()
model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)
agent = create_agent(
    model=model,
    middleware=[
        CallsLimitMiddleware(max_calls=2),
        MessagesLimitMiddleware(max_messages=1),
        LoggingMiddleware(),
    ],  # before_model in order, and after_model in reverse
    checkpointer=InMemorySaver(),
    state_schema=CustomState,
    system_prompt="You are an assistant who answers concisely and clearly.",
)
config = {"configurable": {"thread_id": "conversation_1"}}

contents = [
    "Hello! I am John Steve.",
    "Can you tell me my name?",
    "what's the first question I asked you?",
]
for content in contents:
    response = agent.invoke(
        {"messages": [HumanMessage(content=content)]},
        config=config,
    )
    if "Model call limit reached" in response["messages"][-1].content:
        print(">>> Stopped due to model call limit.")
        break

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents.middleware import SummarizationMiddleware

load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)
agent = create_agent(
    model=model,
    system_prompt="You are an assistant who answers concisely and clearly.",
    middleware=[
        SummarizationMiddleware(
            model=model,  # auto summarize
            trigger=("messages", 6),
            keep=("messages", 2),
            trim_tokens_to_summarize=200,
            # summary_prompt=
        )
    ],
    checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "conversation_1"}}


contents = [
    "Hello! I am John Steve Take Z.",
    "Can you tell me my name?",
    "Can you summarize our conversation so far?",
    "What is my name?",
]
for content in contents:
    response = agent.invoke(
        {"messages": [HumanMessage(content=content)]}, config=config
    )
    print([msg.pretty_print() for msg in response["messages"]])

from langchain.agents.middleware import HumanInTheLoopMiddleware

agent = create_agent(
    model=model,
    tools=[send_email],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"send_email": True}
        )
    ]
)

from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model=model,
    middleware=[
        PIIMiddleware("email", strategy="redact"),   
        PIIMiddleware("phone_number", strategy="block")
    ]
)

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from pydantic import BaseModel, Field, field_validator, ValidationError
from typing import Optional, List
from enum import Enum


# Pydantic Model → JSON Schema as input prompt and output validation
class Priority(str, Enum):
    """person priority levels"""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


class Person(BaseModel):
    """
    A person with a name age and occupation.
    """

    name: str = Field(description="The person's name", min_length=2, max_length=20)
    age: int = Field(description="The person's age", ge=0, le=150)
    occupation: str = Field(description="The person's occupation")
    email: Optional[str] = Field(
        None, description="The person's email address, optional"
    )
    priority: Priority = Field(
        description="The person's priority level: low/medium/high",
        pattern="Priority\.(?:LOW|MEDIUM|HIGH)",
    )
    company_id: int = 321

    @field_validator("email")
    @classmethod
    def validate_email(cls, v, info):
        "custom email validator"
        # info.data include others
        if v is not None and "@" not in v:
            raise ValueError("Invalid email address")
        return v


class PersonList(BaseModel):
    """
    A list of persons.
    """

    persons: List[Person] = Field(description="A list of person objects")


# try:
#     p_invalid = Person(
#         name="Li Si",
#         age=28,
#         occupation="Product Manager",
#         email="lisiatexample.com",  # Invalid email
#         priority=Priority.HIGH,
#         company_id=123,
#     )
# except ValidationError as e:
#     print("Validation Error:", e)
#     exit(1)


load_dotenv()

model = init_chat_model(
    model_provider="openai",
    base_url="https://api.siliconflow.cn/v1/",
    model="deepseek-ai/DeepSeek-R1-0528-Qwen3-8B",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=100,
)
structured_model = model.with_structured_output(PersonList)

prompt = """
Zhang San is a 30-year-old software engineer, medium level priority.
Wu Guang is a 25-year-old data scientist, low level priority, his email is wu.guang@qq.com.
Li Si is a 28-year-old product manager, high level priority.
"""
try:
    response = structured_model.invoke(
        [
            SystemMessage(content="Get Person List from the following text."),
            HumanMessage(content=prompt.strip()),
        ]
    )
    print("return type:", type(response))
    for i, p in enumerate(response.persons):
        print(f"Person {i+1}:")
        print("  name:", p.name)
        print("  age:", p.age)
        print("  occupation:", p.occupation)
        print("  email:", p.email)
        print("  priority:", p.priority)
except ValidationError as e:
    print("Validation Error:", e)