最终章：从脚本到服务 —— 你的AI应用上线之路

本章将简要介绍LangServe，展示如何轻松地将你用LangChain构建的链或代理，打包成一个可供调用的REST API，为你的AI应用走向生产环境铺平道路。

杨小扩

1129人浏览 · 2025-06-11 18:00:00

杨小扩 · 2025-06-11 18:00:00 发布

你好，我是阿扩。恭喜你坚持到了最后！这最后一章，我们将完成从“开发者”到“服务提供者”的思维转变。

在前面的所有章节中，我们的AI应用都以Python脚本的形式，安静地运行在你的本地终端里。这对于学习和开发来说非常棒，但如果你想让你的朋友、同事，甚至全世界的用户都能使用你的AI助理，你需要将它变成一个稳定、可访问的服务（Service）。

今天，我们将学习如何使用LangChain生态中的一个利器——LangServe，轻松地将我们精心打造的AI应用，打包成一个标准的REST API，为它走向生产环境铺平道路。

11.1 本节目标

理解为什么需要将AI应用部署为服务。
了解LangServe的核心优势：自动生成API端点、提供交互式文档页面等。
学会如何修改我们的毕业项目代码，使其能够通过LangServe进行部署。
能够使用uvicorn启动一个本地API服务器，并通过浏览器或API工具（如Postman）与之交互。

11.2 核心概念速览

LangServe：你的AI应用专属“打包发布机”

想象一下，你精心制作了一个美味的蛋糕（你的AI应用）。

本地运行：就像你自己在厨房里品尝这个蛋糕。
部署为服务：就像你开了一家蛋糕店。你需要一个漂亮的展示柜台（API端点），让顾客可以点单；还需要一份清晰的菜单（API文档），告诉顾客有哪些口味、如何点单。

LangServe就是这样一台全自动的“蛋糕店开店机”。你只需要把你的蛋糕（任何LCEL构建的Runnable对象）放进去，它就会自动帮你完成所有事情：

生成API端点：为你的应用自动创建标准的HTTP API接口（如 /invoke, /stream, /batch）。
生成交互式UI：自动创建一个网页界面（我们称之为Playground），你可以在浏览器上直接与你的AI应用进行聊天和测试，非常直观。
生成API文档：自动生成符合OpenAPI规范的API文档（通常在 /docs 路径下），方便前端开发者或其他服务来调用你的AI服务。
处理并发：内置了对高并发请求的处理能力，这是生产环境所必需的。

简单来说，LangServe让你能够专注于AI逻辑本身，而将所有繁琐的Web服务开发工作都自动化了。

11.3 LangChain组件解析

今天我们的主角是LangServe，它主要通过几个简单的函数和命令来工作。

langserve.add_routes:
- 这是LangServe的核心函数。它接收一个FastAPI应用实例、一个你的LCEL链（Runnable对象），以及一个path（API的访问路径），然后自动将所有必要的API路由（endpoints）添加到FastAPI应用中。
FastAPI:
- 它不是LangChain的组件，而是Python世界中一个非常流行、高性能的Web框架。LangServe构建于FastAPI之上，并与其无缝集成。我们只需要了解如何创建一个FastAPI应用实例即可。
Uvicorn:
- 它也不是LangChain的组件，而是一个ASGI服务器。你可以把它看作是运行FastAPI应用的“引擎”或“Web服务器软件”。我们通过uvicorn命令来启动我们的服务。

11.4 实战代码演练

我们将改造第十章的毕业项目，让它从一个命令行交互脚本，变成一个可以通过网络访问的API服务。

第一步：安装必要的库

我们需要安装fastapi, uvicorn 和 langserve。

pip install fastapi uvicorn langserve sse-starlette pydantic

sse-starlette是支持流式输出（Server-Sent Events）所必需的。

第二步：创建服务文件

创建一个名为 assistant_server.py 的新文件。我们将把第十章的代码进行重构，分离出核心的AI逻辑，并用LangServe将其暴露为API。

# assistant_server.py

# 1. 导入必要的模块
import os
from typing import List

# --- 导入所有AI逻辑相关的模块 ---
from langchain_community.chat_models import ChatTongyi
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import tool
from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# --- 导入Web服务相关的模块 ---
from fastapi import FastAPI
from langserve import add_routes
from pydantic import BaseModel # 使用标准的Pydantic V2 BaseModel

# --- 模块一：AI逻辑的构建 (与第十章基本相同) ---

# (请将第十章 personal_assistant_agent.py 中的所有函数和工具定义复制到这里)
# --- START: 从第十章复制的代码 ---
def create_vector_store():
    if os.path.exists("my_knowledge_faiss_index"):
        embeddings = DashScopeEmbeddings(model="text-embedding-v1")
        return FAISS.load_local("my_knowledge_faiss_index", embeddings, allow_dangerous_deserialization=True)
    
    print("构建新的FAISS索引...")
    loader = TextLoader("my_knowledge.md", encoding="utf-8")
    documents = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=250, chunk_overlap=30)
    split_docs = text_splitter.split_documents(documents)
    embeddings = DashScopeEmbeddings(model="text-embedding-v1")
    vector_store = FAISS.from_documents(split_docs, embeddings)
    vector_store.save_local("my_knowledge_faiss_index")
    print("FAISS索引构建并保存成功。")
    return vector_store

llm = ChatTongyi(temperature=0)
vector_store = create_vector_store()
retriever = vector_store.as_retriever(search_kwargs={"k": 2})

rag_prompt_template = """
请根据以下上下文来回答问题。如果上下文中没有相关信息，请明确说明你不知道，不要编造答案。

上下文:
{context}

问题:
{question}
"""
rag_prompt = ChatPromptTemplate.from_template(rag_prompt_template)
rag_chain_base = ({"context": retriever, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser())

@tool
def local_knowledge_searcher(query: str) -> str:
    """
    当用户的问题与LangChain、LCEL或本地知识库中的内容相关时，使用此工具。
    它会查询本地知识库并返回相关信息。
    输入应该是用户的原始问题。
    """
    return rag_chain_base.invoke(query)

@tool
def calculator(expression: str) -> str:
    """
    当需要进行数学计算时，请使用此工具。
    它接收一个字符串形式的数学表达式，并返回计算结果。
    例如: '2+3*5'
    """
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算出错: {e}"

search_tool = DuckDuckGoSearchRun()
search_tool.name = "internet_search"
search_tool.description = "当需要查询实时信息、本地知识库未知的内容或一般性问题时，使用此工具进行网络搜索。"

tools = [local_knowledge_searcher, calculator, search_tool]
agent_prompt = hub.pull("hwchase17/react-chat")
agent = create_react_agent(llm, tools, agent_prompt)
base_agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

personal_assistant = RunnableWithMessageHistory(
    base_agent_executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)
# --- END: 从第十章复制的代码 ---


# --- 模块二：使用FastAPI和LangServe发布服务 ---

# 2. 创建FastAPI应用实例
app = FastAPI(
    title="个人知识助理API服务",
    version="1.0",
    description="一个集成了本地知识、计算、网络搜索和记忆能力的智能助理。",
)

# 3. 定义API的输入模型
# AgentExecutor的输入是一个包含'input'键的字典。
class AgentInput(BaseModel):
    input: str

# 4. 使用add_routes将我们的AI助理添加到FastAPI应用中
add_routes(
    app,
    personal_assistant.with_types(input_type=AgentInput), # 明确指定输入类型
    path="/assistant", # API的访问路径
)

# --- 模块三：(可选) 添加一个根路径用于健康检查 ---
@app.get("/")
def read_root():
    return {"message": "欢迎来到个人知识助理API！访问 /assistant/playground 尝试交互。"}

# --- 模块四：使用uvicorn启动服务 (在命令行中执行) ---
if __name__ == "__main__":
    import uvicorn
    if not os.path.exists("my_knowledge.md"):
        with open("my_knowledge.md", "w", encoding="utf-8") as f:
            f.write("# LangChain\nLangChain是一个框架。\n# LCEL\nLCEL是LangChain表达式语言。")
    
    print("启动API服务，请访问 http://127.0.0.1:8000")
    uvicorn.run(app, host="127.0.0.1", port=8000)

第三步：启动服务

打开你的终端，在assistant_server.py文件所在的目录下，运行以下命令：

uvicorn assistant_server:app --reload

assistant_server: 你的Python文件名（不带.py）。
app: 在该文件中创建的FastAPI实例的变量名。
--reload: 一个方便的开发选项，当你修改代码并保存后，服务器会自动重启。

第四步：与你的服务交互

服务启动后，打开你的浏览器，访问以下地址：

API主页: http://127.0.0.1:8000
- 你会看到我们定义的欢迎信息。
交互式API文档: http://127.0.0.1:8000/docs
- 你会看到一个由Swagger UI生成的专业API文档页面，里面详细列出了所有可用的API端点、参数和返回格式。
LangServe Playground: http://127.0.0.1:8000/assistant/playground/
- 这是最酷的部分！ 你会看到一个聊天界面。你可以在这里直接与你的个人知识助理进行多轮对话。它会自动处理session_id，让你能体验到带记忆的完整功能。尝试问它之前测试过的各种问题，你会发现它的行为和在命令行中完全一致，但体验却友好得多。

11.5 代码逻辑详解

代码重构：我们将AI逻辑的构建部分和Web服务的发布部分清晰地分离开来。AI的核心personal_assistant（一个Runnable对象）被完整地保留了下来。
app = FastAPI(...): 创建了一个Web应用的基础。
class AgentInput(BaseModel): 我们为Agent的输入定义了一个Pydantic模型。这有助于LangServe和FastAPI更好地理解和验证API的输入数据。
add_routes(...): 这是LangServe的魔法所在。
- app: 告诉LangServe要把路由添加到哪个FastAPI应用上。
- personal_assistant.with_types(input_type=AgentInput): 我们把最终的、带记忆的Agent链传给了LangServe。通过.with_types()明确指定输入模型，是一个好习惯。
- path="/assistant": 定义了我们这个AI服务的基础URL路径。所有的相关API都会在这个路径下，例如 /assistant/invoke, /assistant/stream, /assistant/playground。