手把手实现一个「社区旧物回收助手」AI Agent涵盖 RAG 检索、工具调用、上门下单、对话页面基于 FastAPI LangChain 1.x ChromaDB Redis 全栈方案。一、项目背景社区旧物回收场景需要一个智能助手能完成三件事回答回收知识—— 居民问旧衣物怎么回收从知识库检索答案查询回收站—— 居民问朝阳区有哪些回收站查数据库返回开放中的站点预约上门回收—— 居民提供地址和重量创建上门回收订单技术选型组件选型说明Web 框架FastAPI异步高性能自带接口文档AI 框架LangChain 1.xcreate_agent构建 AgentLLM通义千问 qwen-plus阿里云 DashScopeOpenAI 兼容接口向量数据库ChromaDB轻量级本地向量库对话/订单存储Redis对话历史 进行中订单管理ORMTortoise ORM异步 ORM配合 FastAPI二、整体架构用户对话 │ ▼ FastAPI /p4/chat 接口 │ ├── Redis 取对话历史 │ ▼ LangChain create_agent (Agent) │ ├── system_prompt社区旧物回收助手角色 │ ├── 工具1: rag_search ──→ ChromaDB 向量检索 ├── 工具2: search_stations ──→ MySQL 查回收站表 ├── 工具3: create_pickup_order ──→ Redis 存订单 校验 │ ▼ AI 回复 → 存历史 → 返回前端 │ ▼ 对话页 HTML气泡展示三、向量知识库准备在构建 Agent 之前先把回收指南文档向量化存入 ChromaDB。这一步在p42.py中完成import os import chromadb from openai import OpenAI ​ ​ def read_md(): 读取回收指南 markdown 文件 with open(rD:\project\llm\社区旧物回收指南.md, encodingutf-8) as f: return f.read() ​ ​ def split_md(data): 按 200 字符切片 return [data[i:i 200] for i in range(0, len(data), 200)] ​ ​ def embeing_list(list_md): 调用通义千问 embedding 模型生成向量 client OpenAI( api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, ) if isinstance(list_md, str): list_md [list_md] list1 [] batch_chunks 20 chunks [list_md[i:i batch_chunks] for i in range(0, len(list_md), batch_chunks)] for batch in chunks: completion client.embeddings.create( modelqwen3.7-text-embedding, inputbatch, dimensions256 ) list1.extend([item.embedding for item in completion.data]) return list1 ​ ​ def create_chroma(list1, list_md): 向量入库到 ChromaDB client chromadb.PersistentClient(pathrD:\project\p4_b) collection client.get_or_create_collection( namep4_b_chroma_collection, embedding_functionNone # 我们自己生成向量不用内置的 ) ids [fchunk{i} for i in range(len(list1))] collection.add( idsids, embeddingslist1, documentslist_md ) print(fAdded {len(ids)} chunks) ​ ​ if __name__ __main__: data read_md() list_md split_md(data) list1 embeing_list(list_md) create_chroma(list1, list_md)几个关键点split_md切片时用data[i:i200]返回文本块而不是range()返回整数下标这是个容易踩的坑embedding_functionNone表示 ChromaDB 不自动生成向量我们用外部 API 生成后直接传入embeing_list支持批量处理每批 20 个文本块避免 API 超限四、三个工具函数实现Agent 的核心是工具。LangChain 1.x 用tool装饰器定义工具函数签名自动生成 LLM 可见的 schema。4.1 rag_search —— RAG 知识库检索7分from langchain.tools import tool from llm.p42 import embeing_list import chromadb ​ tool(description查询社区旧物回收指南。当居民询问回收分类规则、回收流程、注意事项等知识时使用。参数query: 居民的问题) def rag_search(query: str) - str: # 用 p42.py 的 embeing_list 生成查询向量 query_embedding embeing_list(query) # 返回 [embedding_vector] ​ client chromadb.PersistentClient(pathrD:\project\p4_b) collection client.get_collection( namep4_b_chroma_collection, embedding_functionNone ) res collection.query( query_embeddingsquery_embedding, n_results3 ) docs res[documents][0] return \n.join([f- {d} for d in docs])实现思路用户提问 → 生成查询向量 → ChromaDB 余弦相似度检索 → 返回 top3 文档片段。这里复用了p42.py中的embeing_list函数保证查询向量和入库向量使用相同的模型和维度。4.2 search_stations —— 查询开放中的回收站4分tool(description按区域查询开放中的回收站。当居民想知道某个区域有哪些回收站时使用。参数area: 区域/片区名称) async def search_stations(area: str) - str: stations await Recycle_station.filter( areaarea, status1, is_deleteFalse ).prefetch_related(category) ​ if not stations: return f区域「{area}」暂无开放中的回收站 ​ result [] for s in stations: result.append(f站点{s.name}分类{s.category.name}开放时间{s.open_time}) return \n.join(result)实现思路通过 Tortoise ORM 查询recycle_station表过滤条件status1开放中is_deleteFalse未删除prefetch_related预加载外键关联的分类信息。注意这里是async函数LangChain 工具支持异步。4.3 create_pickup_order —— 创建上门回收订单3分这个工具最复杂涉及两个业务校验和一个巧妙的参数注入技巧from langchain_core.runnables import RunnableConfig import redis import json import time ​ r redis.Redis(hostlocalhost, port6379, db6, decode_responsesTrue, protocol2) ​ tool(description创建上门回收订单。需要提供地址、预估重量(kg)、回收类别。重量限制1~20kg同一用户同时只能有1个进行中订单。参数address: 上门地址, estimated_weight: 预估重量(kg), recycle_category: 回收类别) async def create_pickup_order( address: str, estimated_weight: float, recycle_category: str, config: RunnableConfig # ← 关键自动注入不暴露给 LLM ) - str: # 从 config 中获取 user_id user_id config[configurable][user_id] ​ # 校验1重量 1~20kg if estimated_weight 1 or estimated_weight 20: return f下单失败预估重量必须在 1~20kg 之间当前为 {estimated_weight}kg ​ # 校验2同一用户进行中订单最多1个 active_key fp4:pickup:active:{user_id} if r.exists(active_key): return 下单失败您已有1个进行中的回收订单请等待完成后再下单 ​ # 创建订单存入 Redis order_id fPU{int(time.time())}{user_id} order { order_id: order_id, user_id: user_id, address: address, estimated_weight: estimated_weight, recycle_category: recycle_category, status: pending, create_time: time.strftime(%Y-%m-%d %H:%M:%S) } r.set(active_key, json.dumps(order)) ​ return f上门回收订单创建成功单号{order_id}地址{address}预估重量{estimated_weight}kg回收类别{recycle_category}核心技巧config 参数注入这里有个 LangChain 1.x 的重要特性config: RunnableConfig参数会被自动从工具 schema 中排除LLM 看不到它但我们在调用时可以通过config传入用户上下文。为什么要这样做因为创建订单需要user_id但user_id不应该由 LLM 生成或填充——它应该从登录态/会话中自动获取。通过config注入既保证了安全性LLM 无法伪造 user_id又简化了工具的参数定义。验证一下 LLM 实际看到的工具 schemaprint(create_pickup_order.args) # 输出: {address: {type: string}, estimated_weight: {type: number}, recycle_category: {type: string}} # 注意没有 user_idconfig 被自动排除了五、create_agent 构建 Agent5.1 system_prompt 设计5分SYSTEM_PROMPT 你是「社区旧物回收助手」帮助居民处理旧物回收相关问题。 ​ 你可以使用以下工具 1. rag_search —— 查询社区旧物回收指南分类规则、回收流程、注意事项等知识库内容 2. search_stations —— 按区域查询当前开放中的回收站名称、分类、开放时间 3. create_pickup_order —— 为居民创建上门回收订单需提供地址、预估重量、回收类别 ​ 工作规则 - 居民问回收知识/分类规则时先调 rag_search 查指南再回答 - 居民问附近回收站时调 search_stations 查开放中的站点 - 居民要预约上门回收时调 create_pickup_order 创建订单 - 上门回收重量限制 1~20kg同一居民同时只能有 1 个进行中订单 - 回答要简洁友好用中文 system_prompt 中明确告诉 LLM有哪些工具、每个工具什么时候用、业务规则是什么。这直接影响 Agent 的工具选择准确率。5.2 create_agent 调用8分from langchain_openai import ChatOpenAI from langchain.agents import create_agent ​ TOOLS [rag_search, search_stations, create_pickup_order] ​ def build_agent(): llm ChatOpenAI( modelqwen-plus, temperature0.7, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, ) return create_agent( modelllm, toolsTOOLS, system_promptSYSTEM_PROMPT, ) ​ # 模块级构建import 时创建不会立即调用 LLM recycle_agent build_agent()注意事项create_agent返回的是CompiledStateGraphLangGraph 编译后的图不是旧版 LangChain 的AgentExecutor它没有.tools属性所以暴露工具列表时要从原TOOLS列表取p4_router.post(create_agent, summary创建回收助手Agent) async def create_agent_api(): return { code: 1, message: Agent创建成功, agent: 社区旧物回收助手, tools: [t.name for t in TOOLS], # 从 TOOLS 取不是 agent.tools }模块级构建在 import 时执行create_agent本身不会调用 LLM只是编译图结构所以不会阻塞启动六、对话接口实现6.1 对话主接口from langchain_core.messages import HumanMessage, AIMessage from fastapi import Body p4_router.post(chat, summary对话) async def chat(user_id: int Body(...), query: str Body(...)): # 1. 从 Redis 取最近10条对话历史 history_key fp4:chat:history:{user_id} history_raw r.lrange(history_key, -10, -1) messages [] for h in history_raw: msg json.loads(h) if msg[role] user: messages.append(HumanMessage(contentmsg[content])) else: messages.append(AIMessage(contentmsg[content])) messages.append(HumanMessage(contentquery)) # 2. 调用 Agent通过 config 注入 user_id 给 create_pickup_order result await recycle_agent.ainvoke( {messages: messages}, config{configurable: {user_id: user_id}} ) # 3. 取最后一条 AI 消息 ai_message result[messages][-1].content # 4. 存对话历史到 Redis r.rpush(history_key, json.dumps({role: user, content: query})) r.rpush(history_key, json.dumps({role: assistant, content: ai_message})) return { code: 1, message: 成功, data: {reply: ai_message} }对话流程解析取历史—— 从 Redis List 取最近 10 条消息转成 LangChain 的HumanMessage/AIMessage拼接当前问题—— 把用户最新输入追加到消息列表末尾调用 Agent——ainvoke异步调用config中注入user_id提取回复——result[messages][-1]是 Agent 最后一条回复存历史—— 把本轮 user assistant 消息存回 Redisconfig 注入的传递链路chat 接口 config{configurable: {user_id: user_id}} │ ▼ agent.ainvoke(...) │ ▼ create_pickup_order 工具被调用时 config: RunnableConfig ← 自动接收到 user_id user_id config[configurable][user_id]6.2 查看对话历史p4_router.get(chat/messages/{user_id}, summary查看对话历史) async def get_messages(user_id: int): history_key fp4:chat:history:{user_id} history_raw r.lrange(history_key, 0, -1) messages [json.loads(h) for h in history_raw] return {code: 1, message: 成功, data: messages}七、对话页实现用 FastAPI 的HTMLResponse直接返回一个内嵌的 HTML 页面不需要单独前端项目from fastapi.responses import HTMLResponse p4_router.get(chat/page, summary对话页, response_classHTMLResponse) async def chat_page(): return HTMLResponse(contentCHAT_PAGE_HTML)HTML 页面核心代码!DOCTYPE html html langzh head meta charsetUTF-8 title社区旧物回收助手/title style * { margin: 0; padding: 0; box-sizing: border-box; } body { font-family: Microsoft YaHei, sans-serif; background: #f0f2f5; height: 100vh; display: flex; flex-direction: column; } .header { background: #4CAF50; color: white; padding: 15px 20px; font-size: 18px; font-weight: bold; } .user-bar { padding: 10px 20px; background: white; border-bottom: 1px solid #e0e0e0; display: flex; align-items: center; gap: 10px; } .chat-area { flex: 1; overflow-y: auto; padding: 20px; } .msg { max-width: 70%; margin-bottom: 15px; padding: 10px 15px; border-radius: 10px; line-height: 1.6; } .msg.user { background: #4CAF50; color: white; margin-left: auto; } .msg.assistant { background: white; color: #333; border: 1px solid #e0e0e0; } .input-area { padding: 15px 20px; background: white; border-top: 1px solid #e0e0e0; display: flex; gap: 10px; } .input-area input { flex: 1; padding: 10px 15px; border: 1px solid #ccc; border-radius: 6px; } .input-area button { padding: 10px 25px; background: #4CAF50; color: white; border: none; border-radius: 6px; cursor: pointer; } .input-area button:disabled { background: #ccc; cursor: not-allowed; } /style /head body div classheader社区旧物回收助手/div div classuser-bar label用户ID/label input typenumber iduserId value1 min1 /div div classchat-area idchatArea div classmsg assistant你好我是社区旧物回收助手可以帮你查询回收指南、查找回收站、预约上门回收。请问有什么可以帮你的/div /div div classinput-area input typetext idinput placeholder输入消息回车发送... οnkeydοwnif(event.keyEnter) send() button idbtn οnclicksend()发送/button /div script async function send() { const input document.getElementById(input); const btn document.getElementById(btn); const userId document.getElementById(userId).value; const query input.value.trim(); if (!query) return; const chatArea document.getElementById(chatArea); // 显示用户消息 const userDiv document.createElement(div); userDiv.className msg user; userDiv.textContent query; chatArea.appendChild(userDiv); input.value ; btn.disabled true; btn.textContent 思考中...; // 调用接口 try { const res await fetch(/p4/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ user_id: parseInt(userId), query: query }) }); const data await res.json(); const aiDiv document.createElement(div); aiDiv.className msg assistant; aiDiv.textContent data.data ? data.data.reply : 出错了 data.message; chatArea.appendChild(aiDiv); chatArea.scrollTop chatArea.scrollHeight; } catch (e) { const errDiv document.createElement(div); errDiv.className msg assistant; errDiv.textContent 请求失败 e.message; chatArea.appendChild(errDiv); } btn.disabled false; btn.textContent 发送; } /script /body /html页面功能顶部绿色标题栏 用户ID输入框中间消息区域用户消息靠右绿色气泡AI 消息靠左白色气泡底部输入框 发送按钮支持回车发送发送时按钮禁用显示思考中...防止重复提交八、测试验证8.1 工具 schema 验证from llm.p4_b import rag_search, search_stations, create_pickup_order, TOOLS print(rag_search args:, rag_search.args) # {query: {type: string}} print(search_stations args:, search_stations.args) # {area: {type: string}} print(create_pickup_order args:, create_pickup_order.args) # {address: {type: string}, estimated_weight: {type: number}, recycle_category: {type: string}} # config 被自动排除 print(tools:, [t.name for t in TOOLS]) # [rag_search, search_stations, create_pickup_order]8.2 RAG 检索验证result rag_search.invoke({query: 旧物怎么分类回收}) print(result) # 输出检索到的指南原文片段8.3 上门下单校验验证import asyncio from llm.p4_b import create_pickup_order, r as redis_client async def test(): cfg {configurable: {user_id: 9999}} # 1. 超重测试 print(await create_pickup_order.ainvoke( {address: 测试小区1栋, estimated_weight: 25, recycle_category: 衣物}, cfg)) # 下单失败预估重量必须在 1~20kg 之间当前为 25.0kg # 2. 正常下单 print(await create_pickup_order.ainvoke( {address: 测试小区1栋, estimated_weight: 5, recycle_category: 衣物}, cfg)) # 上门回收订单创建成功单号PU1234567899999... # 3. 重复下单 print(await create_pickup_order.ainvoke( {address: 测试小区2栋, estimated_weight: 8, recycle_category: 家电}, cfg)) # 下单失败您已有1个进行中的回收订单请等待完成后再下单 # 清理 redis_client.delete(p4:pickup:active:9999) asyncio.run(test())三条校验全部通过。九、踩坑总结坑1split_md 返回整数而非文本# 错误写法 —— 返回 [0, 200, 400, ...] 整数下标 list_md [i for i in range(0, len(data), 200)] # 正确写法 —— 返回 [前200字, 第200-400字, ...] 文本块 list_md [data[i:i 200] for i in range(0, len(data), 200)]整数列表传给 embedding API 会直接报类型校验错误。坑2ChromaDB add 长度不一致# 错误写法 —— 循环里每次只传1个向量但 ids/documents 传全部 for i in list1: collection.add(idsids, embeddingsi, documentslist_md) # 长度不匹配 # 正确写法 —— 一次性传入全部 collection.add(idsids, embeddingslist1, documentslist_md)坑3create_agent 返回值没有 .toolsLangChain 1.x 的create_agent返回CompiledStateGraph不是旧版AgentExecutor没有.tools属性。要获取工具列表从原始TOOLS列表取。坑4config 参数不会被 LLM 看到config: RunnableConfig是 LangChain 自动注入的参数不会出现在工具的 args schema 中。利用这一点可以安全地传递 user_id 等上下文而不暴露给 LLM。十、接口总览端点方法说明对应分值/p4/create_agentPOST创建 Agent返回工具列表85分/p4/chatPOST对话接口—/p4/chat/messages/{user_id}GET查看对话历史—/p4/chat/pageGET对话页 HTML10分启动服务uvicorn main:app --reload浏览器打开http://localhost:8000/p4/chat/page即可体验对话。十一、总结本文完整实现了一个基于 LangChaincreate_agent的社区旧物回收 AI 助手核心包括向量知识库用通义千问 embedding 模型 ChromaDB 构建 RAG 检索三个工具rag_search知识检索、search_stations数据库查询、create_pickup_orderRedis 订单 业务校验config 参数注入利用RunnableConfig安全传递 user_idLLM 不可见对话历史Redis List 存储最近 10 条上下文窗口对话页面纯 HTML/CSS/JS 气泡对话界面无需前端框架LangChain 1.x 相比旧版的关键变化create_agent替代initialize_agent返回 LangGraph 编译图tool装饰器支持config: RunnableConfig自动注入Agent 调用统一用ainvoke({messages: [...]})格式希望这篇文章对你有帮助如果有问题欢迎在评论区交流。