FastAPI + LangChain Agent 从零入门学习笔记

一、环境问题总结（Windows 必看）

1. 常见错误

• pip 不是内部或外部命令 → 系统未正确配置 Python 环境变量

• No module named pip → 调用了 LibreOffice 自带阉割版 Python，无开发能力

• uvicorn 不是命令 → 必须用python -m uvicorn 或 py -3 -m uvicorn

• No module named 'dashscope' → 缺少通义千问依赖，需执行 py -3 -m pip install dashscope

• ImportError: cannot import name 'create_react_agent'/'AgentExecutor' → LangChain 版本过新，函数路径变更，改用 LangGraph 极简 Agent 避免报错

• No module named 'langchain.text_splitter'/'langchain.document_loaders' → LangChain 新版包拆分，需改用 langchain_text_splitters 或 langchain_community.document_loaders

• AttributeError: 'VectorStoreRetriever' object has no attribute 'get_relevant_documents' → 新版 Retriever 方法变更，需用 invoke() 替代

2. 万能启动命令（Windows 稳定版）

# 基础依赖安装
py -3 -m pip install fastapi uvicorn langchain langchain-community python-dotenv

# 通义千问依赖安装（接入国内大模型需执行）
py -3 -m pip install dashscope

# LangGraph 依赖安装（打造极简 Agent 需执行）
py -3 -m pip install langgraph

# RAG 相关依赖（文档加载、拆分、向量库）
py -3 -m pip install langchain-text-splitters faiss-cpu

# 启动服务（必用此命令，避免环境冲突）
py -3 -m uvicorn main:app --reload

二、FastAPI 基础（最简可运行版）

1. 最简代码

from fastapi import FastAPI

app = FastAPI(title="AI Agent 服务")

@app.get("/")
def home():
    return {"message": "✅ FastAPI 启动成功"}

@app.post("/chat")
def chat(query: str):
    return {"你的问题": query, "回复": "FastAPI 运行正常"}

2. 访问地址

• 首页：http://127.0.0.1:8000

• 接口文档：http://127.0.0.1:8000/docs（可直接测试接口）

3. 补充：JSON 请求格式（解决参数报错）

之前发送 JSON 请求报错 Field required，原因是接口默认接收「查询参数」，需修改为「JSON 格式」，代码如下：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="AI Agent 服务")

# 定义 JSON 请求格式
class ChatRequest(BaseModel):
    query: str

# 支持 JSON 请求的聊天接口
@app.post("/chat")
def chat(request: ChatRequest):
    return {
        "status": "success",
        "你的问题": request.query,
        "回复": "FastAPI 运行正常"
    }

@app.get("/")
def home():
    return {"message": "✅ FastAPI 启动成功"}

三、LangChain Agent 核心概念

1. Agent 结构

• LLM 大脑：大模型（国内/OpenAI），负责思考、决策

• Tools 工具：计算器、搜索、数据库、API 等，是 Agent 的「能力延伸」

• Prompt：思考逻辑（如 ReAct），指导 Agent 何时调用工具、如何调用

• AgentExecutor：执行器（循环思考→行动→观察），旧版 LangChain 常用，新版易报错，暂不使用

• 记忆（Memory）：存储多轮对话上下文，让 Agent 记住历史交互内容

• RAG 知识库：将本地文档转化为向量，让 Agent 能检索文档内容回答问题，解决大模型“失忆”问题

2. ReAct 流程

1. Thought（思考）：判断问题是否需要调用工具、是否需要检索文档、是否需要结合历史对话

2. Action（行动/调用工具）：选择合适的工具（计算器）或检索文档，传入参数

3. Observation（观察结果）：获取工具执行结果或文档检索结果

4. 重复直到完成任务，返回最终答案

四、完整可运行示例（分阶段实现）

示例1：FastAPI + 通义千问（国内大模型，稳定跑通）

接入国内大模型（通义千问），实现基础 AI 聊天功能，无报错：

from fastapi import FastAPI
from pydantic import BaseModel
import os

# FastAPI 初始化
app = FastAPI(title="我的 AI Agent 项目")

# 通义千问配置（填入自己的 API Key）
from langchain_community.llms import Tongyi
os.environ["DASHSCOPE_API_KEY"] = "你的千问API"
llm = Tongyi(model="qwen-turbo")

# 定义 JSON 请求格式
class ChatRequest(BaseModel):
    query: str

# 普通 AI 聊天接口（不调用工具）
@app.post("/chat")
def chat(request: ChatRequest):
    # 调用通义千问获取回答
    answer = llm.invoke(request.query)
    return {
        "status": "success",
        "question": request.query,
        "answer": answer
    }

# 首页
@app.get("/")
def home():
    return {"message": "✅ FastAPI + 通义千问 运行成功！"}

示例2：打造第一个真正的 Agent（LangGraph 极简版，100% 无报错）

这是我们刚学会的「极简 Agent」，具备思考、决策、执行能力，可调用计算器工具：

from fastapi import FastAPI
from pydantic import BaseModel
import os
import re

# FastAPI 初始化
app = FastAPI(title="我的第一个 AI Agent")

# 1. 通义千问（LLM 大脑）
from langchain_community.llms import Tongyi
os.environ["DASHSCOPE_API_KEY"] = "你的千问API"
llm = Tongyi(model="qwen-turbo")

# 2. 给 Agent 定义工具（计算器）
def add(a: int, b: int) -> int:
    """加法工具，用于计算两个数字的和"""
    return a + b

def multiply(a: int, b: int) -> int:
    """乘法工具，用于计算两个数字的积"""
    return a * b

# 3. Agent 核心逻辑（思考→决策→执行）
def simple_agent(query: str):
    # 提示词：指导 Agent 判断是否需要调用工具
    prompt = f"""
用户问题：{query}

请严格按照以下规则处理：
1. 如果问题是数学计算（加法、乘法），请返回【工具调用】格式：工具名:add 或 multiply, 参数a:数字, 参数b:数字
2. 如果不是数学计算，直接用自然语言回答问题，不要调用工具。
"""

    # 让 LLM 思考，判断是否需要调用工具
    thought = llm.invoke(prompt)

    # Agent 执行：根据思考结果，决定是否调用工具
    if "工具名:add" in thought:
        # 解析参数（提取数字）
        a = int(re.findall(r"参数a:(\d+)", thought)[0])
        b = int(re.findall(r"参数b:(\d+)", thought)[0])
        result = add(a, b)
        return f"我用加法工具计算结果：{result}"

    elif "工具名:multiply" in thought:
        # 解析参数（提取数字）
        a = int(re.findall(r"参数a:(\d+)", thought)[0])
        b = int(re.findall(r"参数b:(\d+)", thought)[0])
        result = multiply(a, b)
        return f"我用乘法工具计算结果：{result}"

    else:
        # 不需要工具，直接返回回答
        return thought

# 4. Agent 接口（支持 JSON 请求）
class ChatRequest(BaseModel):
    query: str

@app.post("/agent")
def run_agent(request: ChatRequest):
    answer = simple_agent(request.query)
    return {
        "status": "success",
        "question": request.query,
        "answer": answer
    }

# 首页
@app.get("/")
def home():
    return {"message": "✅ 我的第一个 AI Agent 启动成功！"}

测试说明：访问 http://127.0.0.1:8000/docs，调用 /agent 接口，传入以下 JSON 可测试 Agent 能力：

• 计算类问题：{"query": "35乘27等于多少？"} → Agent 调用乘法工具返回结果

• 普通问题：{"query": "你是谁？"} → Agent 直接回答，不调用工具

示例3：给 Agent 增加记忆（多轮对话，记住上下文）

在示例2的基础上，增加全局对话记忆，让 Agent 记住上一轮的提问和回答，支持多轮交互（如“35乘27，再加14”）：

from fastapi import FastAPI
from pydantic import BaseModel
import os
import re

# FastAPI 初始化
app = FastAPI(title="带记忆的 AI Agent")

# 1. 通义千问（LLM 大脑）
from langchain_community.llms import Tongyi
os.environ["DASHSCOPE_API_KEY"] = "你的千问API"
llm = Tongyi(model="qwen-turbo")

# 2. 工具定义（计算器）
def add(a: int, b: int) -> int:
    return a + b

def multiply(a: int, b: int) -> int:
    return a * b

# 3. 全局对话记忆（存储历史交互，格式：[{"user": "问题", "ai": "回答"}]）
chat_history = []

# 4. 带记忆的 Agent 核心逻辑
def memory_agent(query: str):
    # 拼接历史对话（让 Agent 记住上下文）
    history_text = ""
    for item in chat_history:
        history_text += f"User: {item['user']}\nAssistant: {item['ai']}\n"

    # 提示词：增加历史对话上下文，指导 Agent 结合记忆处理问题
    prompt = f"""
历史对话：
{history_text}

当前用户问题：{query}

请严格按照以下规则处理：
1. 先结合历史对话，理解用户当前问题（如用户说“再加14”，需结合上一轮计算结果）
2. 如果问题是数学计算（加法、乘法），返回【工具调用】格式：工具名:add 或 multiply, 参数a:数字, 参数b:数字
3. 如果不是数学计算，直接用自然语言回答，不要调用工具。
"""

    thought = llm.invoke(prompt)
    result = ""

    # 工具执行与结果处理
    if "工具名:add" in thought:
        nums = re.findall(r"\d+", thought)
        a = int(nums[0])
        b = int(nums[1])
        res = add(a, b)
        result = f"计算结果：{res}"
    elif "工具名:multiply" in thought:
        nums = re.findall(r"\d+", thought)
        a = int(nums[0])
        b = int(nums[1])
        res = multiply(a, b)
        result = f"计算结果：{res}"
    else:
        result = thought

    # 保存当前对话到记忆中（更新历史）
    chat_history.append({"user": query, "ai": result})
    return result

# 5. 接口定义（增加记忆清空接口）
class ChatRequest(BaseModel):
    query: str

@app.post("/agent")
def run_agent(request: ChatRequest):
    answer = memory_agent(request.query)
    return {
        "status": "success",
        "question": request.query,
        "answer": answer,
        "chat_history": chat_history  # 可选：返回历史对话，方便查看
    }

# 记忆清空接口（避免记忆过多导致混乱）
@app.post("/clear")
def clear_memory():
    global chat_history  # 调用全局变量
    chat_history = []
    return {"msg": "对话记忆已清空"}

@app.get("/")
def home():
    return {"message": "✅ 带记忆的 AI Agent 启动成功！"}

测试说明：

1. 调用 /agent，传入{"query": "35乘27"}，得到结果 945；

2. 再次调用 /agent，传入 {"query": "再加14"}，Agent 会记住上一轮结果 945，计算得出 959；

3. 调用 /clear，可清空历史记忆，重新开始对话。

示例4：RAG 知识库 + Agent（文档问答 + 工具 + 记忆，终极版）

这是我们最终跑通的版本，结合 RAG 知识库（读取本地 txt 文档）、Agent 工具、多轮记忆，实现“文档问答+数学计算+上下文记忆”三合一功能：

from fastapi import FastAPI
from pydantic import BaseModel
import os
import re

# LangChain RAG 相关（新版正确导入路径）
from langchain_community.llms import Tongyi
from langchain_text_splitters import CharacterTextSplitter
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader

# 初始化 FastAPI
app = FastAPI(title="RAG知识库 + 带记忆AI Agent")

# ===================== 1. 配置通义千问 =====================
os.environ["DASHSCOPE_API_KEY"] = "你的千问API"  # 替换为自己的 API Key
llm = Tongyi(model="qwen-turbo")
embeddings = DashScopeEmbeddings()  # 通义千问嵌入模型，用于文档向量化

# ===================== 2. 加载本地文档、构建向量库 =====================
# 加载本地 txt 文档（需在当前目录创建 info.txt，写入文档内容，如“小明是一名程序员，擅长Python开发”）
loader = TextLoader("info.txt", encoding="utf-8")
documents = loader.load()

# 拆分文档（避免文档过长，方便检索）
text_splitter = CharacterTextSplitter(chunk_size=200, chunk_overlap=20)  # chunk_size：每块最大长度，chunk_overlap：块之间重叠长度
split_docs = text_splitter.split_documents(documents)

# 构建 FAISS 向量库（将拆分后的文档转化为向量，用于快速检索）
db = FAISS.from_documents(split_docs, embeddings)
retriever = db.as_retriever()  # 生成检索器，用于查询相关文档

# ===================== 3. 计算器工具 =====================
def add(a: int, b: int) -> int:
    return a + b

def multiply(a: int, b: int) -> int:
    return a * b

# ===================== 4. 全局对话记忆 =====================
chat_history = []

# ===================== 5. RAG+记忆+工具 核心Agent =====================
def rag_agent(query: str):
    # 1. 检索文档相关内容（根据用户问题，从向量库中找匹配的文档片段）
    docs = retriever.invoke(query)  # 新版 Retriever 用 invoke() 替代 get_relevant_documents()
    doc_context = "\n".join([d.page_content for d in docs])  # 拼接检索到的文档内容

    # 2. 拼接历史对话（结合上下文记忆）
    history_text = ""
    for item in chat_history:
        history_text += f"User: {item['user']}\nAssistant: {item['ai']}\n"

    # 3. 构造 Prompt（指导 Agent 结合文档、记忆、工具处理问题）
    prompt = f"""
Document Context:
{doc_context}

Chat History:
{history_text}

User Question: {query}

Rules:
1. If question is math calculate(add/multiply), output only: multiply(num,num) or add(num,num)
2. If question can be answered from Document Context, answer in Chinese based only on context.
3. Else answer normally in Chinese.
"""
    thought = llm.invoke(prompt)

    # 4. 解析工具调用，执行并返回结果
    result = ""
    if "multiply" in thought:
        nums = re.findall(r"\d+", thought)
        a = int(nums[0])
        b = int(nums[1])
        res = multiply(a, b)
        result = f"计算结果：{res}"
    elif "add" in thought:
        nums = re.findall(r"\d+", thought)
        a = int(nums[0])
        b = int(nums[1])
        res = add(a, b)
        result = f"计算结果：{res}"
    else:
        result = thought  # 文档问答或普通问题，直接返回回答

    # 5. 保存当前对话到记忆
    chat_history.append({"user": query, "ai": result})
    return result

# ===================== 接口定义 =====================
class ChatRequest(BaseModel):
    query: str

# RAG+Agent 核心接口
@app.post("/agent")
def run_agent(request: ChatRequest):
    ans = rag_agent(request.query)
    return {
        "question": request.query,
        "answer": ans
    }

# 记忆清空接口
@app.post("/clear")
def clear_memory():
    global chat_history
    chat_history = []
    return {"msg": "对话记忆已清空"}

# 首页（验证服务是否启动）
@app.get("/")
def home():
    return {"message":"✅ RAG知识库 + 带记忆Agent 运行成功"}

前置准备：在 E:\Project\DEMO\python-demo\1 目录下，创建 info.txt 文件，写入任意文档内容（如“小明是一名AI工程师，从事大模型应用开发，擅长FastAPI和LangChain框架”），用于测试文档问答功能。

测试说明：访问 http://127.0.0.1:8000/docs，调用 /agent 接口，测试3种能力：

1. 文档问答：{"query": "介绍一下小明"} → Agent 检索 info.txt 内容，返回对应回答；

2. 数学计算：{"query": "35乘27等于多少"} → Agent 调用乘法工具，返回 945；

3. 多轮记忆：{"query": "再加14"} → Agent 记住上一轮计算结果，返回 959。

五、LangChain 与 LangGraph 核心区别（关键知识点）

1. 一句话总结（必背）

• LangChain：LLM 应用的「组件库 + 简单线性流程框架」，提供工具、LLM 接口、文档加载等基础组件

• LangGraph：LangChain 公司推出的「高级图工作流引擎」，专门实现复杂 Agent 循环、分支、状态管理，完全兼容 LangChain 组件

• 核心关系：组件用 LangChain，复杂流程（Agent）用 LangGraph，是工业界标准组合

2. 核心对比（直接抄笔记）

维度	LangChain	LangGraph
核心结构	链（Chain）→ 线性流程	图（Graph）→ 循环/分支流程
状态管理	弱（简单记忆）	强（全局状态、持久化）
Agent 能力	简单（易报错，需手写逻辑）	强大（官方标准 ReAct 循环，稳定）
控制流	只能顺序执行	支持循环、分支、并行
项目用途	基础 RAG、LLM 接口、工具定义	正规 Agent、多轮对话、复杂流程编排

六、LangGraph 标准 ReAct Agent

1. 核心优势

• 不用手写正则解析工具参数，LLM 自动识别、自动传参

• 内置 ReAct 循环：思考 → 选工具 → 执行 → 再思考 → 输出答案

• 用官方 @tool 装饰器定义工具，规范且可扩展

• 兼容 LangChain 所有组件（LLM、RAG、向量库等）

• 工业级稳定，后续可无缝添加多轮记忆、RAG 检索等功能

2. 关键注意点（避坑重点）

接入通义千问时，必须用 Chat 版模型（ChatTongyi），不能用普通 LLM 模型（Tongyi）；普通 Tongyi 模型不支持 bind_tools 方法，会报AttributeError 错误。

3. 完整可运行代码（100%无报错）

from fastapi import FastAPI
from pydantic import BaseModel
import os

# 关键：导入通义千问 Chat 模型（支持工具绑定）
from langchain_community.chat_models import ChatTongyi
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent

# FastAPI 初始化
app = FastAPI(title="LangGraph 标准 ReAct Agent")

# 通义千问配置（填入自己的 API Key）
os.environ["DASHSCOPE_API_KEY"] = "你的千问API"
llm = ChatTongyi(model="qwen-turbo")  # 正确使用 Chat 版模型

# ===================== 标准工具定义（@tool 装饰器） =====================
@tool
def add(a: int, b: int) -> int:
    """加法工具，用于计算两个整数的和"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """乘法工具，用于计算两个整数的乘积"""
    return a * b

# 工具列表：将所有定义的工具放入列表，供 Agent 调用
tools = [add, multiply]

# ===================== 创建 LangGraph 标准 Agent =====================
# create_react_agent：官方封装的 ReAct 智能体，内置思考-行动-观察循环
agent_executor = create_react_agent(llm, tools)

# ===================== FastAPI 接口 =====================
class ChatRequest(BaseModel):
    query: str  # 定义请求参数（用户问题）

@app.post("/agent")
def run_agent(request: ChatRequest):
    # 调用 Agent，传入用户问题（固定格式：messages = [("user", 问题)]）
    response = agent_executor.invoke({
        "messages": [("user", request.query)]
    })
    # 提取最终回答：消息列表的最后一条就是 Agent 整理好的答案
    answer = response["messages"][-1].content
    return {
        "question": request.query,
        "answer": answer,
        "status": "success"
    }

# 首页：验证服务是否启动成功
@app.get("/")
def home():
    return {"message": "✅ LangGraph 标准 Agent 启动成功！"}

4. 核心代码讲解（重点记忆）

• 工具定义：用 @tool 装饰器修饰函数，函数文档字符串必须写清楚工具用途（LLM 靠此判断何时调用）；参数和返回值需明确类型，方便 LLM 自动传参。

• Agent 创建：create_react_agent(llm, tools) 是 LangGraph 官方封装的 ReAct 智能体，无需手写循环逻辑，自动完成「思考→工具调用→结果整理」。

• Agent 调用格式：必须传入 {"messages": [("user", 问题)]}，这是 LangGraph 标准输入格式。

• 答案提取：response["messages"][-1].content，Agent 的所有交互记录存在 messages 列表中，最后一条是最终回答。

5. 启动与测试步骤

1. 将上述代码复制到 main.py，替换自己的通义千问 API Key；

2. 执行启动命令：py -3 -m uvicorn main:app --reload；

3. 访问接口文档：http://127.0.0.1:8000/docs；

4. 调用 /agent 接口，传入 JSON 测试：

   • 计算题：{"query": "35乘以27再加14等于多少？"} → Agent 自动调用 multiply、add 工具计算；

   • 普通问题：{"query": "你能做什么？"} → Agent 直接回答，不调用工具。

七、学习路线

1. 环境安装 + FastAPI 基础接口（解决环境报错、参数请求格式问题）

2. 接入国内大模型（通义千问），实现基础 AI 聊天

3. LangChain 工具（Tool）定义与调用（如计算器工具）

4. ReAct Agent 原理（思考→行动→观察流程）

5. 打造极简 Agent（LangGraph 版，避开版本报错）

6. 给 Agent 增加记忆（记住上下文，实现多轮对话）

7. RAG 知识库 + Agent（上传文档，实现文档问答）

8. LangGraph 标准 ReAct Agent（官方正规写法，解决工具绑定报错）