手把手教你用LangChain快速构建AI智能体：从模型调用到记忆管理

from langchain_deepseek import ChatDeepSeek
model = ChatDeepSeek(
    model="...",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    # 其他参数...
)

model = ChatOpenAI(
  model="deepseek-chat",
  api_key=os.getenv("DEEPSEEK_API_KEY"),
  base_url="https://api.deepseek.com",
  temperature=0.7
)

model = ChatOpenAI(
        model="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com",
        temperature=0.7
    )

    messages = [
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": "你是谁"}
    ]

result = model.invoke(messages)
print(result.content)
# 流式输出
result = model.stream(messages)
for chunk in result:
    print(chunk.content)

# 定义工具函数
@tool
def get_weather(destination, date):
    # 实际调用天气 API
    return f"{destination} {date} 天气晴朗"
@tool
def get_attractions(destination):
    return f"{destination} 的热门景点有故宫、颐和园..."
# ... 其他函数

from langchain_core.tools import tool
from typing import Annotated
from pydantic import Field

@tool(name_or_callable="get_weather", description="获取指定城市指定日期的天气信息")
def get_weather(
    destination: Annotated[str, Field(description="城市名称，如‘西安’")],
    date: Annotated[str, Field(description="日期，格式 YYYY-MM-DD，如‘2025-05-20’")]
) -> str:
    return f"{destination} {date} 天气晴朗"

@tool(name_or_callable="get_attractions", description="获取指定城市的景点推荐")
def get_attractions(
    destination: Annotated[str, Field(description="城市名称，如‘北京’")]
) -> str:
    # 实际开发中，这里应调用景点 API
    return f"{city} 的热门景点有：故宫、颐和园、天坛。"

from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent
from langchain_core.tools import tool
import os
# 定义工具（假设已有 get_weather, get_attractions）
# 示例工具函数已在前文定义，此处略
# 初始化模型
model = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0.7
)
# 工具列表
tools = [get_weather, get_attractions]
# 创建智能体
agent = create_agent(model=model, tools=tools)
# 准备输入消息
messages = {
    "messages": [
        {"role": "system", "content": "你是一个有用的助手。"},
        {"role": "user", "content": "我想去北京玩，能帮我看看天气和景点吗？"}
    ]
}
# 调用智能体
result = agent.invoke(messages)
# 输出最终答案
print(result["messages"][-1].content)

from langgraph.checkpoint.memory import InMemorySaver
# ...
agent = create_agent(
    # ...
    checkpointer=InMemorySaver()
)
# 调用时指定 thread_id
config = {"configurable": {"thread_id": "1"}}
result = agent.invoke(input=messages, config=config)

import os
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_chroma import Chroma
from langchain_core.documents import Document
# 初始化向量模型
embeddings = DashScopeEmbeddings(
    model="text-embedding-v4",
    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY")
)
# 初始化向量存储
# persist_directory 指定数据持久化目录
vectorstore = Chroma(
    embedding_function=embeddings,
    persist_directory="./chroma_db",
    collection_name="chat_history"
)

def save_messages(messages: str, user_id: int, session_id: int):
    """
    保存用户输入和AI输出的会话记录
    """
    # 将对话内容封装为 Document 对象，并添加元数据
    doc = Document(
        page_content=messages,
        metadata={"user_id": user_id, "session_id": session_id}
    )
    # 添加到向量数据库
    vectorstore.add_documents([doc])

def load_messages(query: str, user_id: int, session_id: int):
    """
    加载用户输入和AI输出的会话记录
    """
    # 构建过滤条件
    filters = []
    if user_id:
        filters.append({"user_id": user_id})
    if session_id:
        filters.append({"session_id": session_id})

    if len(filters) > 1:
        filter_dict = {"$and": filters}
    elif len(filters) == 1:
        filter_dict = filters[0]
    else:
        filter_dict = None

    # 执行相似度搜索，返回最相关的 top-k 记录
    if filter_dict:
        docs = vectorstore.similarity_search(query=query, k=3, filter=filter_dict)
    else:
        docs = vectorstore.similarity_search(query=query, k=3)

    return docs

@wrap_model_call
def add_long_memory(request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]) -> ModelResponse:
    print(f"Model call: {request}")

    message = request.messages[-1]
    if message.type == "human":
        embedding_docs = load_messages(message.content, config["configurable"]["user_id"],
                                       config["configurable"]["session_id"])
        embedding_message = "\n".join([doc.page_content for doc in embedding_docs])
        request.messages.append(ChatMessage(content=embedding_message, role="system"))
    return handler(request)

middleware=[add_long_memory()]

config: RunnableConfig = {"configurable": {"thread_id": "1", "user_id": 1, "session_id": 1}}
 llm = ChatOpenAI(
      model="deepseek-chat",
      api_key=os.getenv("DEEPSEEK_API_KEY"),
      base_url="https://api.deepseek.com",
      temperature=0.7
  )
  messages = {
      "messages": [
          {"role": "system", "content": "你是一个旅游规划助手。"},
          {"role": "user", "content": user_message}
      ]
  }
  tools = [get_weather, get_attractions]
  agent = create_agent(llm,
                       tools=tools,
                       middleware=[add_long_memory()],
                       checkpointer=InMemorySaver())

  result = agent.invoke(input=messages, config=config)
  ai_message = result["messages"][-1].content
  save_messages(messages=user_message + ai_message, user_id=config["configurable"]["user_id"], session_id=config["configurable"]["session_id"])