在本地运行AI智能体很有用，但将其作为服务暴露出来，才能更方便地构建真正的应用程序。

在本教程中，Darsh 将向你展示如何通过 FastAPI 和 Streamlit 来服务一个多用户 AI 智能体。

你将构建一个本地 LangChain 智能体，添加每个用户的会话记忆，通过 HTTP 流式传输响应，并创建一个简单的聊天界面。

https://freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/…

如何用 FastAPI 和 Streamlit 服务多用户 AI 智能体

来源：https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/ 如何用 FastAPI 和 Streamlit 服务多用户 AI 智能体在本教程中，我将向你展示如何通过 FastAPI 将一个支持多个用户的本地 AI 智能体作为 REST API 提供服务，然后在其上添加一个轻量级的 Streamlit UI。

我们不会通过终端与智能体交互，而是通过 HTTP 暴露它，这样多个用户就可以通过聊天风格的前端界面访问它。每个会话都会维护自己的对话历史和流式响应。

该本地 AI 智能体将使用 LangChain v1、Ollama、Qwen 和 Python 构建，运行在你自己的机器上，并且随时可以集成到更大的应用程序中，无需为每次调用支付模型 API 费用。

目录

• 背景 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-background)
• 什么是 FastAPI？ (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-what-is-fastapi)
• 什么是 Streamlit？ (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-what-is-streamlit)
• 什么是多用户支持？ (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-what-is-multi-user-support)
• 动机与架构 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-motivation-and-architecture)
• 步骤 1：安装 Ollama 并拉取模型 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-step-1-install-ollama-and-pull-the-model)
• 步骤 2：安装 Python 依赖 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-step-2-install-python-dependencies)
• 步骤 3：使用 FastAPI 构建智能体和 API 层 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-step-3-build-the-agent-and-api-layer-with-fastapi)
• 步骤 4：构建 Streamlit UI (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-step-4-build-streamlit-ui)
• 步骤 5：运行后端应用 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-step-5-run-the-backend-app)
• 步骤 6：运行前端应用 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-step-6-run-the-frontend-app)
• 示例输出 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-sample-output)
• 投入生产前需要改进的地方 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-what-to-improve-before-production)
• 结论 (https://www.freecodecamp.org/news/how-to-serve-a-multi-user-ai-agent-with-fastapi-and-streamlit/#heading-conclusion)

背景

许多 AI 智能体最初都是简单的 Python 脚本，在命令行终端中运行。你输入一条消息，智能体做出响应，所有操作都在单个本地会话中完成。

这种设置非常适合开发和测试，但当你想让其他人或其他应用程序与智能体交互时，就会受到限制。

要使 AI 智能体真正有用，我们需要通过一个其他用户可以访问的接口来暴露它。REST API 是实现这一目标的一种实用方式。

要学习本教程，你需要在本机上安装 Ollama。本教程适用于 macOS、Windows 和 Linux。我使用的是搭载 32 GB RAM 的 MacBook Pro，但你可以通过从 Ollama 中选择一个更小的 Qwen 模型，在内存较低的机器上运行。

什么是 FastAPI？

FastAPI (https://github.com/fastapi/fastapi) 是一个用于构建 API 的 Python Web 框架。在本教程中，它为我们提供了一种简单的方式，通过 HTTP 暴露智能体，以便其他应用程序、脚本或服务可以调用它。

FastAPI 非常适合 AI 应用程序，因为它为系统提供了一个清晰的边界。我们在 Python 中定义请求和响应模型，FastAPI 会自动验证它们，并将 HTTP 请求转换为 Python 对象，再将 Python 对象转换回 JSON。它还免费生成交互式 API 文档，并支持异步端点，这对于可能需要更长时间响应的 AI 工作负载非常有用。

什么是 Streamlit？

Streamlit (https://streamlit.io/) 是一个 Python 框架，用于构建轻量级的 Web 界面，只需很少的前端工作。它允许我们使用普通的 Python 代码（而不是 HTML、CSS 和 JavaScript）来创建交互式的基于浏览器的应用程序。

在本教程中，Streamlit 作为瘦客户端位于 FastAPI 后端之上。FastAPI 通过 HTTP 暴露 AI 智能体，而 Streamlit 则为我们提供了一个简单的 UI 来调用该 API 并显示结果。这种分离使得后端可重用，同时让智能体在浏览器中易于使用。

什么是多用户支持？

多用户支持意味着 AI 智能体可以处理来自一个以上用户的请求，同时保持每个用户的会话相互独立。

例如，用户 1 问智能体一个问题，用户 2 问一个不同的问题。智能体应该能够独立地为每个用户记住正确的上下文。如果没有多用户支持，所有用户可能会共享相同的对话状态，这可能导致混乱的响应、错误的记忆或被覆盖的上下文。

动机与架构

将 AI 智能体转换为 API 是在本地构建它之后自然而然的下一个步骤。Python 脚本非常适合实验，但 API 使智能体可重用。而添加多用户支持则使智能体可扩展，以供其他人使用。

为了简单起见，我们将使用一个由 Ollama 和 Qwen 驱动的小型本地智能体。该智能体有两个工具：一个用于检查当前时间，另一个用于计数单词。

FastAPI 通过暴露一个名为 /chat/stream 的端点来提供 HTTP 层。当请求到来时，Pydantic 验证请求，LangChain 处理智能体循环和工具调用，最终答案以流的形式返回。Streamlit 位于该 API 之上，作为前端，向 API 发送请求并显示结果。

展示用户调用 Streamlit UI 的序列图。然后到 FastAPI 层，再到 AI 智能体，最后到 Qwen 和工具调用示例请求：

{ "message": "How many words are in: LangChain makes tool calling easier", "user_id":"123e4567-e89b-12d3-a456-426614174000" }

示例响应：

{ "answer": "There are **5** words in LangChain makes tool calling easier." }

模型通过 Ollama 在本地运行，因此无需为每次调用支付模型 API 费用。

步骤 1：安装 Ollama 并拉取模型

首先，为你的平台安装 Ollama 应用程序。

我们将使用 Qwen 作为聊天模型。我使用的是 qwen3.5:4b。如果你的机器内存较小，可以使用 qwen3.5:0.8b。

ollama pull qwen3.5:4b

步骤 2：安装 Python 依赖

创建一个虚拟环境并安装所需的包：

`` python3 -m venv venv source venv/bin/activate

pip install fastapi uvicorn streamlit requests langchain langchain-core langchain-ollama langgraph ``

本教程需要 LangChain >= 1.0.0。

步骤 3：使用 FastAPI 构建智能体和 API 层

该应用程序有三个主要职责。FastAPI 暴露 HTTP 端点，Pydantic 验证传入的请求数据，LangChain 运行智能体，包括工具调用和短期记忆。

每次请求发送的 user_id 被用作线程标识符，允许检查点程序将每个用户的对话历史分开存储。这种记忆是针对会话的。因此，每个新会话都将拥有自己的记忆。

另一个重要的细节是，智能体只在启动时创建一次，使用 agent = build_agent()。重用相同的智能体实例可以避免为每个请求重建模型和工具列表，这减少了开销并提高了响应时间，同时仍然支持多个用户。

在 /chat/stream 端点内部，后端使用 LangChain 的 stream_events(…， version=“v3”) 来生成流式响应，而不是等待完整的答案一次性返回。FastAPI 然后将该流包装在 StreamingResponse 中，这样前端就可以逐步接收输出。这使得应用程序感觉更加交互式，因为用户可以在生成剩余部分的同时立即开始阅读答案。

总之，这为你提供了一个轻量级的后端，可以验证输入，为每个用户保留单独的记忆，并实时将响应流式传输到 UI。

将以下代码保存为 app.py：

`` from datetime import datetime from uuid import UUID

from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse

from pydantic import BaseModel

from langchain.agents import create_agent from langchain_core.tools import tool from langchain_ollama import ChatOllama from langgraph.checkpoint.memory import InMemorySaver

CHAT_MODEL = “qwen3.5:4b”

SYSTEM_PROMPT = ( “You are a helpful assistant with access to tools for getting the current time “ “and counting words in text. “ “Use tools when needed. If the question does not need a tool, answer directly.” )

—————————–

Request model

—————————–

class ChatRequest(BaseModel): user_id: UUID message: str

—————————–

Tools

—————————–

@tool def current_time() -> str: “”“Return the current local date and time.”“” return datetime.now().strftime(“%Y-%m-%d %H:%M:%S”)

@tool def word_count(text: str) -> int: “”“Count the number of words in a piece of text.”“” return len(text.split())

—————————–

Agent + checkpoint memory

—————————–

Store conversation history in short term memory

checkpointer = InMemorySaver()

def build_agent(): model = ChatOllama(model=CHAT_MODEL, temperature=0) return create_agent( model=model, tools=[current_time, word_count], system_prompt=SYSTEM_PROMPT, checkpointer=checkpointer, )

agent = build_agent()

—————————–

Streaming endpoint

—————————–

app = FastAPI()

@app.post(“/chat/stream”) def chat_stream(req: ChatRequest): def generate(): run = agent.stream_events( { “messages”: [{“role”: “user”, “content”: req.message}], }, config={ “configurable”: { # Keep each user’s short-term memory isolated # by using their user_id as the thread ID. “thread_id”: str(req.user_id), } }, version=“v3”, )

    for message in run.messages:
        for token in message.text:
            yield token

return StreamingResponse(generate(), media_type="text/plain")

``

步骤 4：构建 Streamlit UI

Streamlit 代码为 AI 智能体创建了一个简单的聊天界面，并将每个浏览器会话绑定到一个唯一的 user_id。

当应用首次加载时，它会生成一个 UUID 并存储在 st.session_state 中，然后发送到后端，以便智能体可以将该用户的对话历史与其他用户分开。它还会在会话状态中创建一个 chat_history 列表，这样每次 Streamlit 重新运行脚本时，之前的消息仍然可见。然后，应用循环遍历保存的历史记录，并使用 st.chat_message() 以聊天格式显示每条消息。

当用户通过 st.chat_input() 输入新消息时，应用会立即保存并显示该消息，然后通过 POST 请求将其发送到后端 API http://127.0.0.1:8001/chat/stream，同时附带会话的 user_id。

请求使用 stream=True 进行，这使得响应可以逐步到达，而不是一次性全部到达。当从后端接收到每个文本块时，代码将其附加到 full_answer 中，并更新页面上的占位符，从而产生实时流式效果。一旦响应完成，最终的助手消息将存储在 chat_history 中，以便它在页面上保持为对话的一部分。

将以下内容保存为 streamlit_app.py:

`` import uuid import requests import streamlit as st

API_URL = “http://127.0.0.1:8001/chat/stream”

st.title(“Local AI Agent”)

if “user_id” not in st.session_state: st.session_state.user_id = str(uuid.uuid4())

if “chat_history” not in st.session_state: st.session_state.chat_history = []

Show previous messages

for item in st.session_state.chat_history: with st.chat_message(item[“role”]): st.markdown(item[“content”])

message = st.chat_input(“Enter a message”)

if message: # Save and show user message st.session_state.chat_history.append({“role”: “user”, “content”: message}) with st.chat_message(“user”): st.markdown(message)

# Stream assistant response
full_answer = ""
with st.chat_message("assistant"):
    placeholder = st.empty()

    # Send the reqeust to backend API via POST request
    with requests.post(
        API_URL,
        json={
            "message": message,
            "user_id": st.session_state.user_id,
        },
        stream=True,
    ) as response:
        response.raise_for_status()

        for chunk in response.iter_content(chunk_size=None, decode_unicode=True):
            if chunk:
                full_answer += chunk
                placeholder.markdown(full_answer)

# Save final assistant response
st.session_state.chat_history.append(
    {"role": "assistant", "content": full_answer}
)

``

步骤 5：运行后端应用

使用 Uvicorn 启动服务器：

uvicorn app:app --reload --port 8001

应用程序启动后，打开：

• http://127.0.0.1:8001/
• http://127.0.0.1:8001/docs

/docs 端点由 FastAPI 使用你的 Pydantic 模型自动生成。它提供了一个交互式界面，你可以在其中测试 API，而无需编写任何客户端代码。

由 FastAPI 生成的 API 文档。它包含 /chat/stream 端点和模式你可以直接从 curl 发送请求。在终端中，运行以下命令来调用 AI 智能体的 API 并检查输出：

`` $ curl -X POST http://127.0.0.1:8001/chat/stream
-H “Content-Type: application/json”
-d ‘{“message”:“What time is it?”,“user_id”:“123e4567-e89b-12d3-a456-426614174000”}’

$ curl -X POST http://127.0.0.1:8001/chat/stream
-H “Content-Type: application/json”
-d ‘{“message”:“How many words are in: LangChain makes tool calling easier”,“user_id”:“123e4567-e89b-12d3-a456-426614174000”}’

$ curl -X POST “http://127.0.0.1:8001/chat/stream”
-H “Content-Type: application/json”
-d ‘{“message”:“What is the capital of France?”,“user_id”:“123e4567-e89b-12d3-a456-426614174000”}’ ``

要停止服务器，在终端中按 Ctrl+C。

步骤 6：运行前端应用

在另一个终端中，进入项目目录：

source venv/bin/activate streamlit run streamlit_app.py

这将在