引言

入门一下Langchain创建智能体，标题的功能实现参见文章最后的部分。

1. Chat Models

1.1 入门示例

1. 建议使用dotenv加载api_key环境变量
2. 基础的chat模型：from langchain_openai import ChatOpenAI，四个常用参数：model, openai_api_base, openai_api_key, temperature
   示例：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
import os

if __name__ == "__main__":
    load_dotenv()
    llm = ChatOpenAI(model="deepseek/deepseek-chat-v3-0324", 
    openai_api_key=os.getenv("OPENROUTER_API_KEY"),
    openai_api_base="https://openrouter.ai/api/v1",
	temperature=0)
    while True:
        user_input = input("请输入问题：")
        if user_input == "exit":
            break
        result = llm.invoke(user_input)
        print(result.content)

1.2 三种消息类型

即SystemMessage, HumanMessage, AIMessage
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
注：实际上不止3种，还有ToolMessage等等

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
import os

if __name__ == "__main__":
    load_dotenv()
    llm = ChatOpenAI(model="deepseek/deepseek-chat-v3-0324",
                     openai_api_key=os.getenv("OPENROUTER_API_KEY"),
                     openai_api_base="https://openrouter.ai/api/v1",
                     temperature=0)
    while True:
        user_input = input("请输入问题：")
        if user_input == "exit":
            break
        messages = [
            SystemMessage(content="你是一个专业的设计师，请认真回答用户的问题"),
            HumanMessage(content=user_input)
        ]
        result = llm.invoke(messages)
        print(result.content)

效果：

1.3 保存历史对话

用一个list保存上述三种消息

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
import os

if __name__ == "__main__":
    load_dotenv()
    llm = ChatOpenAI(model="deepseek/deepseek-chat-v3-0324",
                     openai_api_key=os.getenv("OPENROUTER_API_KEY"),
                     openai_api_base="https://openrouter.ai/api/v1",
                     temperature=0)
    system_message = SystemMessage(content="你是一个专业的AI助手，请认真回答用户的问题")
    chat_history = [system_message]
    while True:
        user_input = input("请输入问题：")
        if user_input == "exit":
            break
        chat_history.append(HumanMessage(content=user_input))
        result = llm.invoke(chat_history)
        chat_history.append(AIMessage(content=result.content))
        print(result.content)

效果：

2. Prompt Templates

Prompt模板，可以设置替换参数，略，后续有空补

3. Chain

Langchain的链式设计，有点像命令行的管道，略，后续有空补

4. RAG

Langchain的关键功能，略，后续有空补

5. Agents and Tools

5.1 使用FastMCP创建MCP Server

示例：通过经纬度获取天气

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP
from loguru import logger


mcp = FastMCP("get_weather", log_level="ERROR")
API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-agent/1.0"


async def make_nws_request(url: str) -> dict[str, Any] | None:
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None


@mcp.tool()
async def get_weather(latitude: float, longitude: float) -> str:
    """Get the weather forecast for a given latitude and longitude.
    Args:
        latitude(float): The latitude of the location to get the weather forecast for.
        longitude(float): The longitude of the location to get the weather forecast for.
    Returns:
        str: A string representation of the weather forecast.
    """
    # logger.add("weather.log")
    logger.info(f"Getting weather forecast for latitude: {latitude}, longitude: {longitude}")
    url = f"{API_BASE}/points/{latitude},{longitude}"
    data = await make_nws_request(url)
    if not data:
        return "Failed to fetch weather data"
    forecast_url = data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)
    if not forecast_data:
        return "Failed to fetch detailed weather forecast"
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:
        forecast = f"""
{period['name']}:
Temperature: {period['temperature']} {period['temperatureUnit']}
Wind Speed: {period['windSpeed']} {period['windDirection']}
Forecast: {period['detailedForecast']}
"""
        forecasts.append(forecast)
    return "\n---\n".join(forecasts)


if __name__ == "__main__":
    mcp.run(transport="stdio")

注：

1. 使用FastMCP创建mcp对象，第一个参数为mcp服务的名称，log_level建议设置为"ERROR"，防止引起不必要的错误
2. 创建的Function工具，函数前加上@mcp.tool()修饰器即可，然后添加Docstring，这个部分非常重要，将会是提供给LLM识别工具作用的重要参考，另外函数名称、输入参数名称也要明确含义
3. 主函数运行"mcp.run()"，transport指定传输协议，本地用stdio即可，网络部署可以用sse

5.2 使用Langchain/Langgraph创建ReAct Agent

示例：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
import os
import asyncio


system_prompt = """

你是一个AI助手，专门处理天气查询。

"""


async def main():
    load_dotenv()
    llm = ChatOpenAI(model="deepseek-v3-250324",
                     openai_api_base="API Base Url，建议从dotenv加载",
                     openai_api_key="API Key，同上",
                     temperature=0)
    client = MultiServerMCPClient(
        {
            "get_weather": {
                "command": "uv",
                "args": ["run", "./get_weather/main.py"],   #这里是启动命令
                "transport": "stdio"
            }
        }
    )
    tools = await client.get_tools()
    agent = create_react_agent(llm, tools)
    system_message = SystemMessage(content=system_prompt)
    chat_history = [system_message]       #保存上下文
    while True:
        user_input = input("请输入问题：")
        if user_input == "exit":
            break
        user_message = HumanMessage(content=user_input)
        chat_history.append(user_message)
        result = await agent.ainvoke({"messages": chat_history})
        print(result["messages"][-1].content)
        chat_history = result["messages"]
        # print("chat_history", chat_history)


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

注：

1. 使用异步IO启动
2. MultiServerMCPClient可以用于接入多个MCP Server，建议使用
3. agent.ainvoke是异步函数，调用前添加await修饰，返回值的message包含之前的上下文，最后一个为最终答复的AIMessage

效果：