Langchain和fastapi组合，可以构建一个强大的 AI 大模型后端框架，

• FastAPI 是 Web 后端框架：负责处理 HTTP 请求、路由、并发、API 文档等标准后端任务。

• LangChain 是 AI 应用框架：负责编排与大模型的交互、管理提示词、检索数据、调用工具等 AI 专属任务

ps：完整代码放在文章末尾，运行结果可以通过Swagger UI查看。

导入所需要的类和工具

import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict

# 导入LangChain 组件
from config import OPENAI_API_KEY
from langchain_openai import ChatOpenAI
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory

稍微解释一下

• HTTPException 用于返回错误

• CORSMiddleware是一个中间件，它的核心作用就是让不同服务器（不同源）上的前端和后端能够正常通信

• MessagesPlaceholder是一个占位符，用于对多条消息进行占位

初始化fastapi

app = FastAPI(title="猫娘 AI 助手后端")

# 解决跨域问题
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境建议改为具体的域名
    allow_methods=["*"],
    allow_headers=["*"],
)

这里allow_origins后面的可以改成具体的生产环境，比如特定的请求头、HTTP方法；除此之外，有了这个，你可以自己基于vue框架写一个前端文件实现前后端分离的应用。

Agent搭建

首先初始化模型，这里用的是国产模型deepseek，可以根据调用的模型使用相应的api-key

注意：api-key属于重要信息，应该放在config.py配置文件中再导入主程序。

llm = ChatOpenAI(
    model='deepseek-chat',
    base_url='https://api.deepseek.com',
    api_key=OPENAI_API_KEY,
)

定义工具

# 定义工具
@tool
def get_weather(location: str):
    '''获取天气信息'''
    return f'{location}今天天气晴朗，适合出门玩耍喵~'


tools = [get_weather]

这里的工具比较简单，我们模拟了一个获取天气消息的工具函数，实际上复杂一点的工具函数可以调用向量数据库从而获取信息。

注意：这里工具函数中的文本字符串是不可少的，目的是告诉llm这个工具的作用

定义prompt

# 定义提示词
prompt = ChatPromptTemplate.from_messages([
    ('system', '你是一个智能助手，并且可以通过调用工具解决问题'),
    MessagesPlaceholder(variable_name='history'),
    ('user', '{input}'),
    MessagesPlaceholder(variable_name='agent_scratchpad'),
])

• ChatPromptTemplate是专门负责生成提示词模板

• ‘system’表示系统，‘user’表示用户，这个列表就是Openai要求的对话格式；system后面的字符串就是对模型输入的提示词prompt，你可以根据你的需要修改这个字符串，比如你可以改成‘你是一只猫娘‘

• MessagesPlaceholder是一个占位符，variable_name='history'表示插入历史聊天记录，variable_name='agent_scratchpad'非常重要，它负责实现模型工具调用功能。

构建agent

agent = create_tool_calling_agent(llm=llm, prompt=prompt, tools=tools)
agent_executor = AgentExecutor(agent=agent, tools=tools)

# 内存历史记录存储
store: Dict[str, ChatMessageHistory] = {}


def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

先通过create_tool_calling_agent封装agent，再通过AgentExecutor让agent能够自主调用工具

这里store: Dict[str, ChatMessageHistory] = {}简单地创建了一个字典来存储历史聊天记录，实际开发过程中可以替换成数据库。

get_session_history函数负责根据不同的用户id（session_id），来获取对应的聊天记录

带记忆的agent

agent_with_memory = RunnableWithMessageHistory(
    runnable=agent_executor,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="history",  # 务必确保与 Prompt 中的变量名一致
)

最后这里我们使用RunnableWithMessageHistory函数对原来的agent进行封装让agent具有“记忆功能”，我们完成一个简单的、具有记忆、调用工具能力的agent。

定义协议接口

class ChatRequest(BaseModel):
    message: str
    session_id: str


class ChatResponse(BaseModel):
    output: str
    session_id: str

这里我们定义两个数据模型，用于规范 FastAPI 接口的请求体和响应体的格式；告诉前端发来的数据应该长什么样，后端返回的数据又该长什么样

所以ChatRequest在客户端发送 POST 请求时使用，ChatResponse在服务器返回数据时使用

​​​​​​定义API路由

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        # 调用 LangChain 逻辑
        result = agent_with_memory.invoke(
            {"input": request.message},
            config={"configurable": {"session_id": request.session_id}}
        )

        # 返回结果
        return ChatResponse(
            output=result["output"],
            session_id=request.session_id
        )
    except Exception as e:
        # 捕获异常并返回 500 错误
        raise HTTPException(status_code=500, detail=str(e))

这里首先定义了一个post接口，路径为/chat

这个response_model是一个fastapi的关键字，负责指定响应格式模型response_model=ChatResponse 表明返回数据必须符合ChatResponse格式

agent_with_memory.invoke（）是在将输入喂给llm并把输出返回给result，最后返回的result其实是个字典，键“output”的值才是真正的llm返回文本，并把llm返回的内容封装成ChatResponse格式

except Exception as e:表示当 try 块中的代码出错时，捕获异常，然后返回一个标准的 HTTP 500 错误响应给客户端。

完整代码如下：

import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict

# 导入你之前的 LangChain 组件
from config import OPENAI_API_KEY
from langchain_openai import ChatOpenAI
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory


app = FastAPI(title="AI 助手后端")

# 解决跨域问题（让你的 Vue 前端可以访问）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  
    allow_methods=["*"],
    allow_headers=["*"],
)




llm = ChatOpenAI(
    model='deepseek-chat',
    base_url='https://api.deepseek.com',
    api_key=OPENAI_API_KEY,
)



@tool
def get_weather(location: str):
    '''获取天气信息'''
    return f'{location}今天天气晴朗，适合出门玩耍喵~'


tools = [get_weather]

prompt = ChatPromptTemplate.from_messages([
    ('system', '你是一个智能助手，并且你会调用工具解决问题。'),
    MessagesPlaceholder(variable_name='history'),
    ('user', '{input}'),
    MessagesPlaceholder(variable_name='agent_scratchpad'),
])


agent = create_tool_calling_agent(llm=llm, prompt=prompt, tools=tools)
agent_executor = AgentExecutor(agent=agent, tools=tools)


store: Dict[str, ChatMessageHistory] = {}


def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]


agent_with_memory = RunnableWithMessageHistory(
    runnable=agent_executor,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="history",  # 务必确保与 Prompt 中的变量名一致
)



class ChatRequest(BaseModel):
    message: str
    session_id: str


class ChatResponse(BaseModel):
    output: str
    session_id: str



@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
       
        result = agent_with_memory.invoke(
            {"input": request.message},
            config={"configurable": {"session_id": request.session_id}}
        )

        
        return ChatResponse(
            output=result["output"],
            session_id=request.session_id
        )
    except Exception as e:
       
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == '__main__':
    uvicorn.run(app, host="127.0.0.1", port=8000)