在本教程中，我将向您展示如何使用FastAPI将多用户本地AI代理作为REST API提供出来，然后再为其添加一个轻量级的Streamlit用户界面。
我们不会通过终端与这个代理进行交互，而是通过HTTP将其暴露出来，这样多个用户就可以通过类似聊天界面的前端来访问它。每个会话都会保留自己的对话历史记录以及接收到的响应内容。
这个本地AI代理是使用LangChain v1、Ollama、Qwen以及Python构建的，它可以在您的个人电脑上运行，而且可以轻松地集成到更大的应用程序中，使用时也不会产生任何每次调用模型时所需的API费用。
目录
背景知识
许多AI代理最初都是简单的Python脚本，它们在命令行终端中运行。用户输入信息后，代理会做出响应，所有这些操作都发生在同一个本地会话中。
这种设置非常适合开发和测试阶段，但当需要其他人员或应用程序与这个代理进行交互时，这种架构就会变得不够灵活了。
要想让AI代理真正发挥作用，我们就需要通过一个其他人能够访问的接口来暴露它。而REST API正是实现这一目标的理想方式。
要学习本教程，您需要在自己的电脑上安装Ollama。本教程适用于macOS、Windows和Linux系统。我使用的是一台配备32GB内存的MacBook Pro，但如果您使用的内存较少，也可以选择Ollama中内存需求较低的Qwen模型来运行本教程。
什么是FastAPI？
FastAPI是一个用于构建API的Python Web框架。在本教程中，它为我们提供了一种简单的方法，通过HTTP将AI代理暴露出来，这样其他应用程序、脚本或服务就可以调用它了。
FastAPI非常适合用于开发AI应用程序，因为它能为我们的系统提供清晰的边界。我们可以在Python中定义请求和响应模型，FastAPI会自动对这些模型进行验证，并能将HTTP请求转换为Python对象，再将Python对象转换回JSON格式。此外，FastAPI还能免费生成交互式的API文档，同时还支持异步端点，这对于那些响应时间可能较长的AI应用来说非常有用。
什么是Streamlit？
Streamlit是一个Python框架，它允许我们用最少的前端开发工作来构建轻量级的Web界面。通过使用普通的Python代码，而不是HTML、CSS和JavaScript，我们可以创建出交互式的浏览器应用程序。
在这个教程中，Streamlit作为一层薄客户端运行在FastAPI后端之上。FastAPI通过HTTP接口暴露AI代理功能，而Streamlit则为我们提供了简单的用户界面，用于调用这些API并显示结果。这种分离机制使得后端代码可以重复使用，同时也能让AI代理在浏览器中方便地被使用。
什么是多用户支持？
多用户支持意味着AI代理能够处理来自多个用户的请求，并且能够保持每个用户的会话状态相互独立。
例如，用户1向代理提出一个问题，而用户2又提出了另一个问题。代理应该能够分别为每个用户记住正确的上下文信息。如果没有多用户支持，所有用户可能会共享相同的对话状态，从而导致响应混乱、数据错误或上下文被覆盖等问题。
开发动机与架构设计
在本地构建完AI代理之后，将其转化为API是自然而然的下一步。虽然Python脚本非常适合用于实验目的，但API能让这个代理具备更高的复用性。而加入多用户支持功能后，这个代理就能被其他人用来进行实际应用了。
为了简化开发流程，我们将使用由Ollama和Qwen支持的简单本地代理。这个代理包含两个功能模块：一个用于查看当前时间，另一个用于统计单词数量。
FastAPI通过提供名为/chat/stream的接口来实现HTTP层的功能。当用户发送请求时，Pydantic会验证请求内容，LangChain会负责处理代理逻辑及相关工具的调用，最终答案会以流的形式返回。Streamlit则作为前端框架，负责向这个API发送请求并展示结果。
示例请求格式：
{
"message": "‘LangChain使工具调用更加方便’这句话包含多少个单词？",
"user_id":"123e4567-e89b-12d3-a456-426614174000"
}
示例响应：
{
"answer": "LangChain中有**5**个单词，这些单词使得工具调用变得更加便捷。"
}
该模型通过Ollama在本地运行，因此每次调用该模型时都不会产生任何费用。
步骤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对象中，因此前端可以逐步接收生成的响应内容。这样一来，应用程序会显得更加交互性强，因为用户可以在响应内容还在生成的过程中就开始阅读。
综上所述，这套架构能够实现输入验证、为每个用户保留独立的内存空间，并实时将响应数据推送到用户界面。
请将以下代码保存为app.py文件：
from datetime import datetime
from uuid import UUID
from fastapi import FastAPI, HTTPException
from fastapiresponses 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 = (
"你是一个功能强大的助手，可以使用各种工具来获取当前时间，
也可以统计文本中的单词数量。在需要时可以使用这些工具；如果问题不需要使用工具，可以直接回答。"
)
# -----------------------------
# 请求模型
# -----------------------------
class ChatRequest(BaseModel):
user_id: UUID
message: str
# -----------------------------
# 工具函数
# -----------------------------
@tool
def current_time() -> str:
"""返回当前的本地日期和时间。"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def word_count(text: str) -> int:
"""统计文本中包含的单词数量。"""
return len(text.split())
# -----------------------------
# 代理模型 + 检查点内存机制
# -----------------------------
# 将对话历史记录存储在短期内存中
checkpointer = InMemorySaver()
def build_agent():
model = ChatOllama(model=CHAT_MODEL, temperature=0)
return create_agent(
model=model,
tools=[current_time, word_count],
systemprompt=SYSTEM_PROMPT,
checkpointer=checkpointer,
)
agent = build_agent()
# -----------------------------
# 流式响应端点
# -----------------------------
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": {
# 通过使用用户的 user_id 作为线程 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用户界面
Streamlit代码为AI助手创建了一个简单的聊天界面，并确保每个浏览器会话都与一个唯一的user_id关联。
当应用程序首次加载时，它会生成一个UUID并将其存储在st.session_state中，之后这个UUID会被发送到后端服务器，这样助手就能将该用户的对话记录与其他用户的记录区分开来。同时，代码还会在session状态中创建一个chat_history列表，这样每次重新运行脚本时，之前的消息都会显示出来。应用程序会遍历这些保存下来的历史记录，并使用st.chat_message()函数以聊天格式显示每一条消息。
当用户通过st.chat_input()输入新消息时，应用程序会立即保存并显示这条消息，然后通过POST请求将其与session的user_id一起发送到后端API地址http://127.0.0.1:8001/chat/stream。
这个请求是使用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("本地AI助手")
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 = []
# 显示之前的消息
for item in st.session_state.chat_history:
with st.chat_message(item["role"]):
st.markdown(item["content"])
message = st.chat_input("输入一条消息")
if message:
# 保存并显示用户的消息
st.session_state.chat_history.append({"role": "user", "content": message})
with st.chat_message("user"):
st.markdown(message)
# 显示助手的回复
full_answer = ""
with st.chat_message("assistant"):
placeholder = st.empty()
# 通过POST请求将消息发送到后端API
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)
# 保存助手的最终回复
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的功能。
你也可以直接使用curl发送请求。在终端中运行以下命令，就可以调用AI代理的API并查看输出结果：
$ curl -X POST http://127.0.0.1:8001/chat/stream \
-H "Content-Type: application/json" \
-d '{"message":"现在几点了？","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":"“LangChain让工具调用变得更加方便”这句话包含多少个单词？","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":"法国的首都是什么？","user_id":"123e4567-e89b-12d3-a456-426614174000"}'
要停止服务器，请在终端中按Ctrl+C。
步骤6：运行前端应用
在另一个终端中，进入项目目录：
source venv/bin/activate
streamlit run streamlit_app.py这样，你的浏览器就会打开http://localhost:8501/这个地址，从而访问前端应用。你可以尝试输入像“法国的首都是什么？”这样的问题，然后会看到以聊天界面形式显示的答案。
这个前端界面实际上是在调用FastAPI的端点，并通过这些接口来与AI代理进行交互。现在，你已经拥有一个可以实际使用的、端到端的功能完备的应用程序了。
要停止服务器，请在终端中按Ctrl+C。
示例输出结果
下图显示了在同一端点上运行的两个浏览器会话。每个会话都会被分配一个唯一的标识符，这样后端就可以为每个用户维护独立的对话记录。
即使两个用户都问了“我是谁？”这个问题，得到的回答也会不同，因为每个会话的回答都是基于它之前接收到的信息来生成的。
在生产之前需要改进的地方
尽管这个应用程序已经具备了所有功能，但它仍然被设计得相当简洁。目前它已经支持可重用的FastAPI后端、Streamlit聊天界面、用户个性化的对话记录以及实时响应功能。
如果你想进一步开发这个项目，接下来的步骤包括添加身份验证机制、持久化存储系统、结构化的日志记录功能、监控系统，以及更完善的部署方案。
还需要注意的是，如果你的目标仅仅是快速搭建一个功能完备的自主托管聊天界面，那么你可能并不需要自己从头开始开发前端代码。像LibreChat和Open WebUI这样的工具已经提供了非常丰富且易于使用的界面功能。
本教程采用了另一种方法：它并没有选择使用现成的完整平台，而是展示了如何自己构建一个轻量级的定制开发环境，这样你就能更深入地了解其架构，并能更好地控制这个AI代理的暴露方式。
结论
通过本教程，我们把一个独立的AI代理封装到了一个FastAPI应用程序中，并在其上使用了Streamlit用户界面。
这样的设计使得这个AI代理不再只是一个独立的脚本，而变成了一种可重复使用的服务。现在，其他应用程序、脚本或内部工具都可以通过简单的HTTP接口来访问它。
由于为每个会话分配了唯一的标识符，该服务能够为多个用户分别维护对话记录，从而实现每个会话都拥有独立内存空间的聊天界面。
从这里开始，你可以通过添加身份验证功能或生产环境所需的其他特性来继续扩展这个服务。祝你在开发过程中取得成功！
如果你喜欢本教程，可以在我的博客上找到我更多的文章（最近的文章包括一系列关于系统设计的论文），也可以在我的个人网站上了解我的工作进展，同时还可以在LinkedIn上关注我的动态。
