本文详细介绍了如何使用FastAPI和LangGraph从零构建一个能够使用工具的AI Agent。通过创建计算器、时间查询等工具，结合LangGraph的ReAct模式，实现了能够进行多轮对话并调用工具的智能系统。文章提供了完整的项目结构和代码实现，包括非流式和流式API接口，帮助开发者快速掌握AI Agent开发技术。

准备好了吗？让我们一步一步开始构建吧！

项目结构一览

一个清晰的项目结构是良好开端。我们的项目将组织如下：

agent-demo/├── app/│   ├── agents/│   │   └── basic_agent.py      # LangGraph Agent 的组装逻辑│   ├── api/│   │   └── query_routes.py     # FastAPI 路由层│   ├── core/│   │   ├── config.py           # 应用配置│   │   ├── deps.py             # 依赖注入│   │   └── llm_loader.py       # 大语言模型加载器│   ├── schemas/│   │   └── chat_schema.py      # Pydantic 数据模型│   ├── services/│   │   └── agent_service.py    # Agent 调用服务│   ├── tools/│   │   └── demo_tools.py       # Agent 可用的工具│   ├── lifespan.py             # 应用生命周期管理│   └── main.py                 # 应用主入口├── .env                        # 环境变量└── pyproject.toml              # 项目与依赖管理

第一步：安装依赖

首先，让我们来安装所有需要的 Python 库。我们将使用 uv 这个现代化的包管理工具，当然你也可以使用 pip。

uv add fastapi fastapi-cli uvicorn loguru pydantic-settings langchain-core langchain-ollama langgraph ollama gradio

第二步：为 Agent 打造“工具箱”

一个 Agent 的强大之处在于它能够使用工具。让我们先来定义几个简单的工具函数。

app/tools/demo_tools.py:

import randomfrom datetime import datetimefrom langchain_core.tools import tool, ToolException# --- 1. 定义工具函数 ---@tooldef calculator(expression: str) -> str:    """一个安全的计算器函数。当需要进行数学计算时使用。"""    try:        # 安全性检查：只允许特定的字符        allowed_chars = "0123456789+-*/(). "        if not all(char in allowed_chars for char in expression):            return "错误: 输入包含不允许的字符。"        # 使用安全的 eval 执行        result = eval(expression, {"__builtins__": None}, {})        return str(result)    except Exception as e:        raise ToolException(f"计算出错: {e}")@tooldef get_current_time() -> str:    """获取当前日期和时间字符串。"""    current_time = datetime.now().isoformat()    return f"当前时间是{current_time}"@tooldef dice_roller(sides: int = 6) -> str:    """掷一个指定面数的骰子并返回结果。"""    try:        result = random.randint(1, sides)        return str(result)    except Exception as e:        return f"掷骰子出错: {e}"# --- 2. 将工具函数聚合到一个列表中 ---tools = [calculator, get_current_time, dice_roller]

代码释义:

• • @tool 装饰器: 这是 LangChain 的“魔法棒”。将它放在一个函数上，LangChain 就能自动理解这个函数的用途（通过函数的文档字符串 """..."""）、输入参数和返回类型。这样，大语言模型（LLM）就能知道在何时、如何调用这个工具。
• • 文档字符串 (Docstring): """...""" 里的内容至关重要！LLM会读取这段描述来决定是否使用这个工具。描述越清晰，Agent 的决策就越准确。
• • tools 列表: 我们将所有定义好的工具函数放在一个列表中，方便后续统一提供给 Agent。

第三步：配置我们的 FastAPI 应用

接下来，我们来设置一些“老朋友”——配置文件、LLM 加载器等。

app/core/config.py:

from functools import lru_cachefrom pydantic_settings import BaseSettings, SettingsConfigDictclass Settings(BaseSettings):    """应用主配置"""    # 应用基本信息    app_name: str = "Agent Service Demo"    debug: bool = False    # Ollama 配置    ollama_base_url: str = "http://localhost:11434"    ollama_default_model: str = "qwen3:4b-instruct"    # pydantic-settings 的配置    model_config = SettingsConfigDict(        env_file=".env", env_file_encoding="utf-8", case_sensitive=False    )@lru_cachedef get_settings() -> Settings:    """返回一个缓存的 Settings 实例，确保配置只被加载一次。"""    return Settings()settings = get_settings()

模型选择提示: 在这里我使用了 qwen3:4b-instruct 模型。它小巧、高效，遵循指令的能力很强，非常适合这个项目。经验表明，至少需要 3B 参数量以上的模型才能较好地驱动 Agent 使用工具。

app/core/llm_loader.py:

from functools import lru_cachefrom loguru import loggerfrom langchain_ollama import ChatOllamafrom app.core.config import settings@lru_cache(maxsize=1)def get_llm() -> ChatOllama:    """    根据全局配置初始化并返回一个 ChatOllama 实例。    这是一个同步函数，将在应用的 lifespan 中被调用。    """    model = settings.ollama_default_model    logger.info(f"正在初始化Ollama LLM: {model}...")    try:        llm = ChatOllama(            model=model, base_url=settings.ollama_base_url, temperature=0.5        )        logger.success(f"Ollama LLM '{model}' 初始化成功。")        return llm    except Exception as e:        logger.error(f"初始化 Ollama LLM 失败: {e}")        raise

代码释义:

• • @lru_cache(maxsize=1): 这是一个非常实用的 Python 装饰器。maxsize=1 意味着它会缓存第一次调用的结果。当 get_llm() 被再次调用时，它会立即返回缓存中的同一个 LLM 实例，而不会重新创建一个新的。这保证了在整个应用中，我们使用的是同一个、唯一的 LLM 对象，避免了资源浪费。
• • 职责分离: 这个文件的作用是集中管理 LLM 的创建。如果未来我们想换成 OpenAI 或其他模型，只需要修改这一个文件，而不需要改动应用的其他部分。

第四步：重头戏——组装 LangGraph Agent

现在，让我们来构建 Agent 的“大脑”。

app/agents/basic_agent.py:

from langchain_core.runnables import Runnablefrom langchain_ollama import ChatOllamafrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.prebuilt import create_react_agentfrom loguru import loggerfrom app.tools.demo_tools import toolsasync def assemble_langgraph_agent(llm: ChatOllama) -> Runnable:    """    一个“组装厂”函数，负责将 LLM 和工具组装成一个可运行的 LangGraph Agent。    """    logger.info("开始组装 LangGraph Agent...")    # MemorySaver 用于在多次调用之间保持对话状态（短期记忆）。    # 对于生产环境，可以替换为 SqliteSaver, RedisSaver 等持久化存储。    memory = MemorySaver()    # create_react_agent 是 LangGraph 提供的一个高级工厂函数，    # 它可以快速创建一个遵循 ReAct (Reason + Act) 逻辑的 Agent。    runnable_agent = create_react_agent(        model=llm,        tools=tools,        checkpointer=memory,    )    logger.success(f"✅ LangGraph Agent 组装完成，使用模型: {llm.model}")    return runnable_agent

代码释义:

这段代码是 Agent 的核心组装逻辑，可以理解为一个“Agent 组装厂”。

1. 1. MemorySaver(): 这是 Agent 的“短期记忆”。它能让 Agent 记住同一场对话中的上下文。LangGraph 提供了多种 checkpointer（检查点），如 SqliteSaver、RedisSaver，可以将对话历史持久化到数据库中，实现长期记忆。
2. 2. create_react_agent(...): 这是 LangGraph 提供的“一键组装”功能。它会创建一个遵循 ReAct (Reasoning + Acting) 模式的 Agent。

• • Reasoning (思考): LLM 会根据你的问题和可用工具进行思考，决定下一步该做什么。例如，“用户问现在几点，我应该使用 get_current_time 工具。”
• • Acting (行动): LLM 决定调用工具，然后 LangGraph 框架会实际执行这个工具函数。
• • 这个过程会循环进行，直到 LLM 认为它已经获得了足够的信息来回答你的初始问题。

3. 3. 参数说明:

• • model=llm: 告诉 Agent 使用哪个 LLM 作为“大脑”。
• • tools=tools: 将我们之前定义的工具列表提供给 Agent。
• • checkpointer=memory: 为 Agent 配置好记忆系统。

最终，这个函数返回一个 runnable_agent，这是一个可以被调用和执行的完整 Agent 对象。

第五步：应用生命周期管理 (Lifespan)

我们使用 FastAPI 现代化的 lifespan state 模式，在应用启动时加载模型和 Agent，并在请求处理中安全地访问它们。

app/lifespan.py:

from typing import TypedDictfrom collections.abc import AsyncIteratorfrom contextlib import asynccontextmanagerfrom fastapi import FastAPIfrom loguru import loggerfrom langchain_ollama import ChatOllamafrom langchain_core.runnables import Runnablefrom app.core.config import get_settingsfrom app.core.llm_loader import get_llmfrom app.agents.basic_agent import assemble_langgraph_agentclass AppState(TypedDict):    """定义应用生命周期中共享状态的结构。"""    llm: ChatOllama    agent: Runnable@asynccontextmanagerasync def lifespan(app: FastAPI) -> AsyncIterator[AppState]:    # -------- 启动 --------    get_settings()    logger.info("🚀 应用启动,配置已就绪。")    # 1. 加载 LLM    llm = get_llm()    # 2. LLM健康检查    try:        await llm.ainvoke("Hello")        logger.success("✅ LLM 健康检查通过...")    except Exception as e:        logger.error(f"❌ LLM 初始化失败: {e}")        raise RuntimeError("LLM 初始化失败，应用启动中止")    # 3. 组装 LangGraph Agent    agent = await assemble_langgraph_agent(llm)    # 4. 通过 yield 将状态传递给应用    yield AppState(llm=llm, agent=agent)    # -------- 关闭 --------    logger.info("应用关闭，资源已释放。")

代码释义:

lifespan 就像是应用的“开/关机”流程。

• • yield 之前: 这部分代码在 FastAPI 应用启动时执行。我们按顺序：加载配置 -> 初始化 LLM -> 对 LLM 进行一次快速的“健康检查” -> 组装 Agent。
• • 健康检查: await llm.ainvoke("Hello") 这一步非常重要。它确保了我们的应用在启动时就能连接到 Ollama 服务，如果连接失败，应用会立即报错退出，而不是等到第一个用户请求进来时才发现问题。
• • yield AppState(...): 这是关键。yield 将 llm 和 agent 实例作为一个字典“产出”。FastAPI 会将这个字典保存在 app.state 中。这样，在后续的请求中，我们可以通过 request.state 来安全地访问这些已经初始化好的对象。
• • yield 之后: 这部分代码在应用关闭时执行，用于进行资源清理。

第六步：服务与依赖注入

为了在路由函数中方便地获取 Agent 和 LLM，我们创建依赖注入函数。

app/core/deps.py:

# app/core/deps.pyfrom typing import castfrom fastapi import Request, HTTPException, statusfrom langchain_core.runnables import Runnablefrom langchain_ollama import ChatOllamadef get_agent(request: Request) -> Runnable:    """    一个 FastAPI 依赖项。    它从 lifespan state (request.state) 中获取 Agent 实例。    这是 FastAPI 推荐的、类型安全的方式。    """    if not hasattr(request.state, "agent"):        raise HTTPException(            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,            detail="Agent service is not initialized."        )    # 使用 cast 帮助类型检查器理解 request.state.agent 的确切类型    return cast(Runnable, request.state.agent)def get_llm(request: Request) -> ChatOllama:    """    一个 FastAPI 依赖项，用于从 request.state 中获取 LLM 实例。    """    if not hasattr(request.state, "llm"):        raise HTTPException(            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,            detail="LLM service is not initialized."        )    return cast(ChatOllama, request.state.llm)

现在，我们来编写调用 Agent 的核心服务逻辑，支持流式和非流式两种方式。

app/services/agent_service.py:

# app/services/agent_service.pyimport jsonfrom typing import AsyncGeneratorfrom loguru import loggerfrom langchain_core.runnables import Runnable, RunnableConfigfrom langchain_core.messages import HumanMessage, AIMessageasync def run_agent(agent: Runnable, query: str, session_id: str) -> str:    """    以非流式的方式异步调用 Agent，并传入会话ID。    """    logger.info(f"非流式调用 Agent [Session: {session_id}], 查询: '{query}'")    config: RunnableConfig = {"configurable": {"thread_id": session_id}}    result = await agent.ainvoke({"messages": [HumanMessage(content=query)]}, config)    for message in reversed(result["messages"]):        if isinstance(message, AIMessage):            content = message.content            if isinstance(content, str):                logger.success(f"成功获取非流式响应 [Session: {session_id}]。")                return content            elif isinstance(content, list):                text_parts = [                    part["text"]                    for part in content                    if isinstance(part, dict) and part.get("type") == "text"                ]                if text_parts:                    return "\n".join(text_parts)    raise ValueError("Agent 未能生成有效的 AI 响应。")async def stream_agent(    agent: Runnable, query: str, session_id: str) -> AsyncGenerator[str, None]:    """    以流式的方式异步调用 Agent，并传入会话ID。    """    logger.info(f"流式调用 Agent [Session: {session_id}], 查询: '{query}'")    config: RunnableConfig = {"configurable": {"thread_id": session_id}}    logger.debug(f"准备调用 Agent.astream_events，传入的 config: {config}")    # 1. 模型流式输出    async for event in agent.astream_events(        {"messages": [HumanMessage(content=query)]}, version="v1", config=config    ):        kind = event["event"]        if kind == "on_chat_model_stream":            chunk = event.get("data", {}).get("chunk")            if chunk:                content = chunk.content                if content:                    yield f"data: {json.dumps({'content': content})}\n\n"

代码释义:

这部分是实际与 LangGraph Agent 交互的地方。

1. 1. 会话管理 (session_id):

• • config: RunnableConfig = {"configurable": {"thread_id": session_id}}
• • 这是 LangGraph 中实现多轮对话记忆的关键！我们将 session_id 包装在 configurable 字典的 thread_id 字段中。LangGraph 会根据这个 thread_id 去查找对应的对话历史 (checkpointer)，从而实现上下文记忆。

2. 2. 非流式调用 (run_agent):

• • 使用 agent.ainvoke() 发起调用。
• • 它会等待 Agent 完成所有思考和工具调用，最后返回一个包含完整对话历史的结果。
• • 我们的代码需要从返回的消息列表中，找到最后一条 AIMessage，并提取其内容作为最终答案。

3. 3. 流式调用 (stream_agent):

• • 使用 agent.astream_events()。这会返回一个异步事件流。
• • 我们监听 event["event"] == "on_chat_model_stream" 这种类型的事件，这代表 LLM 正在生成内容的“token 流”。
• • 我们将每个内容块 (chunk.content) 包装成 Server-Sent Event (SSE) 格式 (data: {...}\n\n) 并 yield 出去，这样前端就能实时接收并显示了。

第七步：定义 API 接口

我们需要定义请求和响应的数据结构，这得益于 Pydantic 的强大功能。

app/schemas/chat_schema.py:

from pydantic import BaseModel, Fieldclass ChatRequest(BaseModel):    """聊天请求的数据模型。"""    query: str = Field(..., min_length=1, description="用户的输入查询")    session_id: str | None = Field(default=None, description="会话ID，用于保持多轮对话。")class ChatResponse(BaseModel):    """非流式聊天响应的数据模型。"""    answer: str = Field(..., description="Agent 返回的最终答案")    session_id: str = Field(..., description="当前会话的ID。")

为什么需要 Schema?

在 FastAPI 中，使用 Pydantic 模型（Schema）能带来三大好处：

1. 1. 自动数据校验：FastAPI 会自动检查请求体是否符合 ChatRequest 的格式。
2. 2. 自动类型转换：如果可能，它会自动转换数据类型。
3. 3. 自动生成文档：Swagger UI 或 ReDoc 页面会根据这些模型生成清晰、可交互的 API 文档。

最后，我们来编写 FastAPI 路由，将所有部分连接起来。

app/api/query_routes.py:

# app/api/v1/agent.pyimport uuidfrom loguru import loggerfrom fastapi import APIRouter, Depends, HTTPException, statusfrom starlette.responses import StreamingResponsefrom langchain_core.runnables import Runnablefrom langchain_ollama import ChatOllamafrom app.core.deps import get_agent, get_llmfrom app.schemas.chat_schema import ChatRequest, ChatResponsefrom app.services import agent_servicerouter = APIRouter(tags=["Agent"])@router.post("/chat/invoke", response_model=ChatResponse, summary="非流式聊天")async def chat_invoke(    request: ChatRequest,    # 使用 Depends 从 lifespan state 中安全地注入 agent 实例    agent: Runnable = Depends(get_agent)):    """    与 Agent 进行一次性问答。    - 如果请求中不提供 `session_id`，将创建一个新的会话。    - 如果提供 `session_id`，将继续在该会话中进行多轮对话。    """    # 核心会话管理逻辑：如果客户端没有提供，我们在此处生成一个新的    session_id = request.session_id or str(uuid.uuid4())        try:        # 将 agent, query, 和 session_id 一同传递给服务层        answer = await agent_service.run_agent(agent, request.query, session_id)        return ChatResponse(answer=answer, session_id=session_id)    except Exception as e:        logger.error(f"Agent 调用失败 [Session: {session_id}]: {e}")        raise HTTPException(            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,            detail=f"Agent execution failed: {e}"        )@router.post("/chat/stream", summary="流式聊天")async def chat_stream(    request: ChatRequest,    agent: Runnable = Depends(get_agent)):    """    与 Agent 进行流式聊天，支持多轮对话记忆。    """    session_id = request.session_id or str(uuid.uuid4())        # 注意：流式响应通常不直接返回 session_id，客户端需要自己管理。    # 一种常见的做法是客户端在第一次请求后，从非流式端点或特定session端点获取ID。    return StreamingResponse(        agent_service.stream_agent(agent, request.query, session_id),        media_type="text/event-stream"    )

代码释义:

• • Depends(get_agent): 这是 FastAPI 的依赖注入系统。当请求到达时，FastAPI 会自动调用 get_agent 函数，并将返回的 agent 实例注入到我们的路由函数中。这让我们的代码非常干净和解耦。
• • 会话 ID 管理: session_id = request.session_id or str(uuid.uuid4()) 这行代码实现了简单的会话管理。如果客户端在请求中提供了 session_id，我们就使用它；如果没有，就用 uuid.uuid4() 创建一个新的，并返回给客户端，以便它在后续请求中使用。
• • StreamingResponse: 对于流式接口，我们返回 StreamingResponse。它的第一个参数是我们的 stream_agent 生成器函数，FastAPI 会自动处理，将生成的内容以流的形式发送给客户端。media_type="text/event-stream" 明确告诉浏览器这是一个 Server-Sent Event 流。

最后一步：启动应用

记得在 main.py 中挂载你的路由，然后就可以启动应用了！你可以在 FastAPI 自动生成的 Swagger UI (http://127.0.0.1:8000/docs) 中测试你的 Agent。

Gradio 前端聊天界面代码可以在项目仓库中找到，有兴趣的同学可以自行复制或拉取。

总结

恭喜你！通过这个小项目，你已经掌握了：

• • 如何为 AI Agent 创建和注册工具。
• • 使用 LangGraph 的 create_react_agent 快速构建一个具备思考和行动能力的 Agent。
• • 通过 lifespan state 模式在 FastAPI 中安全高效地管理 Agent 和 LLM 实例。
• • 实现了支持多轮对话记忆的流式和非流式 API 接口。

这为你打开了通往更复杂、更有趣的 AI 应用开发的大门。快去动手试试，让你的 Agent 拥有更多强大的工具吧！

读者福利大放送：如果你对大模型感兴趣，想更加深入的学习大模型**，那么这份精心整理的大模型学习资料，绝对能帮你少走弯路、快速入门**

如果你是零基础小白，别担心——大模型入门真的没那么难，你完全可以学得会！

👉 不用你懂任何算法和数学知识，公式推导、复杂原理这些都不用操心；
👉 也不挑电脑配置，普通家用电脑完全能 hold 住，不用额外花钱升级设备；
👉 更不用你提前学 Python 之类的编程语言，零基础照样能上手。

你要做的特别简单：跟着我的讲解走，照着教程里的步骤一步步操作就行。

包括：大模型学习线路汇总、学习阶段，大模型实战案例，大模型学习视频，人工智能、机器学习、大模型书籍PDF。带你从零基础系统性的学好大模型！

现在这份资料免费分享给大家，有需要的小伙伴，直接VX扫描下方二维码就能领取啦😝↓↓↓

为什么要学习大模型？

数据显示，2023 年我国大模型相关人才缺口已突破百万，这一数字直接暴露了人才培养体系的严重滞后与供给不足。而随着人工智能技术的飞速迭代，产业对专业人才的需求将呈爆发式增长，据预测，到 2025 年这一缺口将急剧扩大至 400 万！!

大模型学习路线汇总

整体的学习路线分成L1到L4四个阶段，一步步带你从入门到进阶，从理论到实战，跟着学习路线一步步打卡，小白也能轻松学会！

大模型实战项目&配套源码

光学理论可不够，这套学习资料还包含了丰富的实战案例，让你在实战中检验成果巩固所学知识

大模型学习必看书籍PDF

我精选了一系列大模型技术的书籍和学习文档(电子版)，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

大模型超全面试题汇总

在面试过程中可能遇到的问题，我都给大家汇总好了，能让你们在面试中游刃有余

这些资料真的有用吗？

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

👉获取方式：

😝有需要的小伙伴，可以保存图片到VX扫描下方二维码免费领取【保证100%免费】

相信我，这套大模型系统教程将会是全网最齐全 最适合零基础的！！