当前位置：首页 > news >正文

使用OpenAI、LangChain、MongoDB构建一个AI agent

news 来源：原创 2025/7/20 17:02:03

像 ChatGPT、Gemini 和 Claude 这样的大模型已成为企业必不可少的工具。如今，几乎每家公司都希望根据自己的需求或客户群体，开发一款定制化的AI Agent。

这篇文章将重点介绍如何创建一个个性化的助手，这个助手不仅能进行功能调用，还能将对话记录存储在数据库中，实现多会话的连续互动，同时能够执行网页搜索并总结相关信息。

为了更好地组织结构并便于未来扩展功能，使用 Langchain、Langgraph 和 LangSmith 这三个工具，它们会简化整个过程，并提升系统的功能性。Langchain 可以简化流媒体处理、工具调用，并支持多种不同的 LLM。Langgraph 是一个组织工具，帮助我们选择使用的工具，并且让智能助手自主决策路径。LangSmith 则是一个观察工具，它能帮助我们监控向 LLM 提问到最终得到答案的整个过程。

资源准备

Tavily API key
OpenAI API key
MongoDB Atlas
Langsmith API key

首先，在你的 .env 文件里加上几个关键的环境变量，包括 openai_key、tavily_key 和 mongo_uri。

OPENAPI\_KEY\=OPENAI\_KEY=sk-proj-XXXXXX
TAVILY\_API\_KEY\=tvly-XXXXXXXXXXXXXXXXXXXXXXXX
MONGO\_URI\=mongodb+srvXXXXXXXXXXXXXXXXXXXXXXXXXXXLANGCHAIN\_TRACING\_V2\=true
LANGCHAIN\_ENDPOINT\=https://api.smith.langchain.com
LANGCHAIN\_API\_KEY\=lsv2\_pt\_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\_xxxxxxxxxx
LANGCHAIN\_PROJECT\=project-name

然后，把这些Key和 URI 都配置到你的设置文件里，确保 langsmith、OpenAI key、Tavily key和 Mongo URI 一个都不少。

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDictclass Settings(BaseSettings):openai_key: SecretStrtavily_api_key: strmongo_uri: SecretStr# LangSmithlangchain_tracing_v2: bool = Truelangchain_endpoint: str = "https://api.smith.langchain.com"langchain_project: str = "ai-agent"langchain_api_key: strmodel_config = SettingsConfigDict(env_file=".env", extra="ignore")settings = Settings()

使用设置中的API key设置 OpenAI 客户端。

from langchain_openai import ChatOpenAIfrom .utilities.settings import settingsllm = ChatOpenAI(openai_api_key=settings.openai_key,model_name="gpt-4o-mini",max_retries=2,
)

Langgraph

为了让聊天机器人的各个部分能够顺畅交流，需要设置一个全局“状态”变量，这样数据就能在各个节点间流通了。用 LLM 生成消息之后，通过 langgraph 的 add_messages 函数，把这些消息加入到聊天机器人的记忆里。

from typing import Annotated, TypedDictfrom langchain_core.messages import BaseMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.graph.state import CompiledStateGraphclass State(TypedDict):messages: Annotated[list[BaseMessage], add_messages]

在 langgraph 中，每个节点其实就是一个函数。我们现在要做一个新的 chatbot 函数，它会通过 llm.ainvoke 去调用 OpenAI，把当前的消息状态传过去。OpenAI 会给我们返回一条新的 AI 消息，然后我们就把这个新消息更新到状态里。通过上一步配置的 add_messages 函数，新消息会自动加入到现有的消息队列中。

最后提一下，langgraph 提供了同步和异步两种方法。如果你想要异步处理，只需在方法名前加个 a。这次我们选择异步的 ainvoke，而不是同步的 invoke。

async def chatbot(state: State) -> State:"""Chatbot"""response_message = await llm.ainvoke(state["messages"])return {"messages": [response_message]}

现在，我们要用 add_node 把新写的函数加入到流程图里。在这个图中，节点之间的连线就是所谓的“边”

接下来，要用 add_edge 把图的起点 START 和聊天机器人节点连起来。最后用 graph_builder.compile 来编译这个图表。不编译的话，这个图表是没法用的。

其实这就是代码话的流程编排，很多AI WebUI都有。

async def get_graph() -> CompiledStateGraph:    """    Get the graph    """    graph_builder = StateGraph(State)    graph_builder.add_node("chatbot", chatbot)    graph_builder.add_edge(START, "chatbot")    graph_builder.add_edge("chatbot", END)    graph = graph_builder.compile()    return graph

目前流程图长这样：

图1 流程图

然后让我们来编写一个新异步函数 run_graph，通过 graph.astream（同步的话就用 stream）来启动我们精心编译的图。这个函数会激活“chatbot”节点，也就是之前定义的 chatbot 函数。

首先，它会放入一个System Prompt，再加上用户提出的问题，这样就构成了我们的对话状态。这些信息会被发送到 OpenAI，AI会回一条消息，我们的函数则会把这个回复加入到对话状态中。

在图的循环中，我们可以随时查看事件变量，了解当前的对话状态。我们会从这些事件中挖掘出 value，再从这个 value 中提取出最后一条消息的精华，最后把这份精华传回给用户。

至于系统提示，它就像是为 AI 助手提供了一份“情报”，让它的搜索更加精确。因为 LLM 的训练数据有截止时间，所以它可能会更擅长挖掘历史资料，而不是最新的知识。

async def get_graph() -> CompiledStateGraph:"""Get the graph"""graph_builder = StateGraph(State)graph_builder.add_node("chatbot", chatbot)graph_builder.add_edge(START, "chatbot")graph_builder.add_edge("chatbot", END)graph = graph_builder.compile()return graph

接下来，写一个新异步函数 main，这个函数的核心任务就是接收用户的输入。同时，加入一个循环语句，它的作用是监听用户的输入，一旦用户键入（quit、exit 或 q），聊天就会在终端优雅地结束。

from datetime import datetime, timezone
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessagenow = datetime.now(timezone.utc)system_prompt = f"""
You are an AI assistant helping an user.Current Date: {now}
"""asyncdef run_graph(question: str) -> None:"""Run the graph"""asyncfor event in graph.astream({"messages": [SystemMessage(content=system_prompt),HumanMessage(content=question),]}):for value in event.values():print(value["messages"][-1].content)returnNone

工具调用

现在，我们要利用 .env 和 .settings 文件中的 tavily_api_key。通过 TavilySearchResults，我们的 AI 助手就能在网上进行搜索了。这些搜索工具会被集成到 tools 中，方便未来我们随时扩充工具箱。然后，我们通过 llm_with_tools = llm.bind_tools 这行代码，将工具列表绑定到 AI 助手身上。最后，记得把代码中所有用到 llm 的地方，都替换成 llm_with_tools，这样一来，AI 助手就能使用这些工具了。

async def main() -> None:"""AI Agent"""whileTrue:question = input("q: ")if question.lower() in ["quit", "exit", "q"]:print("Goodbye!")breakawait run_graph(question)returnNoneif __name__ == "__main__":anyio.run(main)

下一步，加入一个 tool_node，这个节点的作用是激活调用 AI 消息的工具。设置一个 add_conditional_edge，这个特殊的边可以让 AI 助手根据不同的需求，导向不同的节点。它会先判断是否需要进行工具调用，如果不需要，那么流程就会直接导向 END 节点，结束这次交互。

from langchain_community.tools.tavily_search import TavilySearchResultsweb_search = TavilySearchResults(max_results=2)
tools = [web_search]llm = ChatOpenAI(openai_api_key=settings.openai_key,model_name="gpt-4o-mini",max_retries=2,
)llm_with_tools = llm.bind_tools(tools)asyncdef chatbot(state: State) -> State:"""Chatbot"""response_message = await llm_with_tools.ainvoke(state["messages"])return {"messages": [response_message]}

存储对话数据

接下来，利用 .env 和 .settings 文件中的 mongo_uri。首先，在代码的顶部加入 async_mongodb_client。然后，我们会在 get_graph 函数中设置一个检查点，它的作用是将记忆存储到 mongoDB。注意，这个操作需要放在异步函数里面才能正常执行哦。

async def get_graph() -> CompiledStateGraph:"""Get the graph"""graph_builder = StateGraph(State)graph_builder.add_node("chatbot", chatbot)tool_node = ToolNode(tools=tools)graph_builder.add_node("tools", tool_node)graph_builder.add_edge(START, "chatbot")graph_builder.add_conditional_edges("chatbot", tools_condition)graph_builder.add_edge("tools", "chatbot")graph = graph_builder.compile()return graph

现在，利用 MongoDB 来设置 thread_id，这样做可以保存不同聊天记录中的消息。这个功能很强大，它让助手能够跨多个聊天 session 记住之前的对话内容。而且，如果想要为不同的聊天创建独立的记录，只需要使用不同的 thread_id 就可以轻松实现。

from typing import Annotated, Any, TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.mongodb.aio import AsyncMongoDBSaver
from motor.motor_asyncio import AsyncIOMotorClientasync_mongodb_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(settings.mongo_uri.get_secret_value()asyncdef get_graph() -> CompiledStateGraph:"""Get the graph"""checkpointer = AsyncMongoDBSaver(client=async_mongodb_client,db_name="ai",checkpoint_collection_name="checkpoints",writes_collection_name="checkpoint_writes",)graph_builder = StateGraph(State)graph_builder.add_node("chatbot", chatbot)tool_node = ToolNode(tools=tools)graph_builder.add_node("tools", tool_node)graph_builder.add_edge(START, "chatbot")graph_builder.add_conditional_edges("chatbot", tools_condition)graph_builder.add_edge("tools", "chatbot")graph = graph_builder.compile(checkpointer=checkpointer)return graph

现在，可以执行脚本，看看运行结果。由于输入的是当前日期，所以它能展示出团队最新的动态和信息。

from langchain_core.runnables import RunnableConfigasyncdef run_graph(config: RunnableConfig, question: str) -> None:"""Run the graph"""graph = await get_graph()asyncfor event in graph.astream({"messages": [SystemMessage(content=system_prompt),HumanMessage(content=question),]},config=config,stream_mode="values",):event["messages"][-1].pretty_print()returnNoneasyncdef main() -> None:"""AI Agent"""config = RunnableConfig(configurable={"thread_id": 1})whileTrue:question = input("q: ")if question.lower() in ["quit", "exit", "q"]:print("Goodbye!")breakawait run_graph(config=config, question=question)returnNone

可以在你的 langsmith 项目中逐步查看运行过程。

在这里插入图片描述
图2 Langsmith 可视化Trace