这次我们来看一个很有意思的实战型项目用 LangChain 多智能体架构加 Python 和 Streamlit做一个“婚礼策划师”应用。它不是概念演示而是把策划过程中的需求分析、预算规划、场地建议、时间线安排、供应商协调这些环节拆成多个智能体让每个智能体负责一个专业角色再由 LangGraph 统一编排最后通过 Streamlit 做出可视化操作界面。整个项目的核心是“多智能体怎么协作”而婚礼策划只是落地场景。这个项目最值得关注的有几点第一门槛不高默认走大模型 API 调用不要求本地有一张高显存显卡中等配置电脑完全能跑第二功能边界清晰每个智能体只做一件事方便调试和扩展后续换场景、加角色都很快第三有完整的 Web 界面不是只能在终端里敲命令适合做演示、接外包原型、或者给非技术用户试用第四可以继续封装成 API 服务也支持批量任务比如一次处理多对新人的策划需求。本文会带你把环境、架构、代码、界面、测试、API 封装、批量任务整个链路跑通。阅读本文需要一点 Python 基础至少知道怎么装依赖、怎么跑脚本。如果你正在学 LangChain 和 LangGraph又不想只看官方文档的抽象例子这个项目会是比较完整的练手素材。下面直接进入正文。1. 多智能体婚礼策划师核心能力速览能力项说明项目类型多智能体应用开发实战LangChain LangGraph Streamlit实现语言Python大模型接入默认使用 OpenAI 兼容接口可以填 DeepSeek、通义千问、Kimi 等平台的 API Key硬件要求调用云端 API 时基本不依赖本地 GPU中等配置电脑即可本地模型支持理论上可以换成 Ollama 等本地推理服务但显存和速度取决于模型大小需自行测试多智能体角色婚礼策划师、预算顾问、场地推荐官、供应商协调员、时间线管理员编排方式LangGraph 状态图各节点按流程依次执行并共享上下文Web 界面Streamlit 聊天式交互支持输入新人需求后自动生成完整方案API 服务可用 FastAPI 封装成 HTTP 接口供其他系统调用批量任务支持读取 CSV 需求列表批量生成策划方案 Markdown 文档适合场景LangChain 学习实战、多智能体入门、策划行业工具原型、企业内部流程演示需要说明一点这里说的“中配”不是指某个特定显卡型号而是指“不依赖本地大模型推理、只要能正常跑 Python 和浏览器”的日常开发环境。如果你要换成本地模型请以实际测试为准。2. 适用场景与使用边界这个项目适合三类人。第一类是正在学 LangChain 的开发者想搞明白 Prompt 模板、模型封装、链式调用、多智能体编排分别解决什么问题婚礼策划这个场景足够具体比“通用助手”更容易理解。第二类是产品原型开发者需要快速搭建一个“多角色协作 可视化界面”的 Demo用来给客户或领导演示。第三类是策划行业的技术爱好者想验证 AI 在方案生成、预算测算、时间线管理上能省多少人工。它不适合什么场景如果你想要的是一个生产级别的智能策划系统那还差得远。大模型存在幻觉生成的场地价格、供应商名单必须人工核实不同地区的婚礼习俗差异很大通用 Prompt 覆盖不了全部情况涉及真实客户姓名、联系方式、预算金额这类信息时直接丢给第三方 API 会有隐私风险。另外大模型 API 服务偶尔会不稳定如果做成线上系统必须有超时重试、降级方案和人工审核环节。合规方面要特别注意不要拿真实客户隐私数据做测试不要用模型生成的报价单直接发给客户不要把未经授权的真实商家信息当作“推荐结果”输出如果以后接入了图片生成、人脸相关的能力还需要确认肖像权和内容授权。技术本身是中性的但使用边界要清楚。3. 整体架构为什么用 LangGraph 编排多智能体多智能体项目最常见的误区是把多个 Prompt 放在一个数组里轮询假装它们是“多个智能体”。真正可维护的多智能体系统至少要满足三点每个智能体只负责一个职能智能体共享会话状态流程可以被控制、被观察、被修改。LangGraph 的状态图正好对应这套设计。本项目的流程是这样设计的用户输入婚礼需求包括预算、人数、日期、风格偏好、所在城市等随后第一个节点是“策划师”负责理解需求并拆解出待办事项之后“预算顾问”根据预算和人数输出分配建议“场地推荐官”结合城市和风格给出场地类型和筛选方向“供应商协调员”生成摄影、化妆、餐饮、主持等品类的注意事项“时间线管理员”负责把以上内容排成婚礼当天的日程。所有节点运行完后由一个汇总节点生成完整的策划方案 Markdown。LangGraph 在这里的价值是它让每个节点之间的“数据流”变得明确。你可以把状态定义为一个字典每个 Agent 从状态里读自己需要的字段再把产出写回去。后续加一个“应急预案智能体”或“宾客管理智能体”只需要新增节点和边不用推翻重写。从工程角度讲这种架构还方便调试。节点执行到哪一步、哪一步耗时最长、哪一步返回了非法 JSON都能通过日志定位。对学习者来说理解这个流程之后再去看 LangChain 官方文档或 LangGraph 示例会轻松很多。后面我们直接用代码把它搭出来。4. 环境准备与前置条件开始之前先确认本机环境满足下面这些条件。检查项建议值说明操作系统Windows 10/11、macOS、Linux 均可本项目的核心代码跨平台Python3.9 及以上建议 3.10 或 3.11LangChain 和 LangGraph 对高版本支持更好pip最新版本避免依赖解析问题API Key一个 OpenAI 兼容的模型服务 Key以 DeepSeek 或通义千问为例按平台要求开通网络能正常访问模型 API 服务具体域名以你选择的平台为准浏览器Chrome / Edge 等现代浏览器用于打开 Streamlit 界面磁盘预留 2GB 以上主要是 Python 依赖和日志模型文件不下载则占用很小如果你的电脑已经装了 Anaconda 或 Miniconda可以用虚拟环境隔离项目。如果没有直接用 Python 自带的 venv 也可以。这一步很重要因为 LangChain、Streamlit、FastAPI 都有各自的依赖混装在全局环境里容易冲突。检查 Python 版本python --version如果版本低于 3.9建议先升级 Python 环境。这里给出一套通用的虚拟环境创建命令实际路径按你自己的项目目录调整mkdir langchain-wedding-planner cd langchain-wedding-planner python -m venv venvWindows 下激活虚拟环境venv\Scripts\activatemacOS / Linux 下激活虚拟环境source venv/bin/activate激活后命令行前面会出现(venv)标记。接下来安装依赖。5. 安装部署与项目初始化在项目根目录下新建requirements.txt写入以下依赖langchain0.3.0 langchain-openai0.2.0 langgraph0.2.0 streamlit1.36.0 python-dotenv1.0.0 fastapi0.111.0 uvicorn0.30.0 pandas2.0.0安装命令pip install -r requirements.txt如果你的网络环境使用镜像源可以临时指定国内 PyPI 镜像例如pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple然后创建.env文件保存模型服务的配置。这里以 OpenAI 兼容接口为例LLM_API_KEYsk-你的密钥 LLM_BASE_URLhttps://api.deepseek.com/v1 LLM_MODELdeepseek-chat LLM_TEMPERATURE0.7如果使用其他平台把LLM_BASE_URL和LLM_MODEL换成你实际的服务地址和模型名。这里只写通用模板具体值以你的模型服务商提供的信息为准。项目目录结构建议这样组织langchain-wedding-planner/ ├── .env ├── requirements.txt ├── agents/ │ ├── __init__.py │ ├── llm.py │ ├── planner.py │ ├── budget.py │ ├── venue.py │ ├── supplier.py │ └── timeline.py ├── graph.py ├── app.py ├── api_service.py ├── batch_run.py ├── outputs/ └── data/ └── requirements.csvagents目录用来存放每个智能体的 Prompt 和节点逻辑graph.py负责用 LangGraph 组装整个多智能体流程app.py是 Streamlit 前端api_service.py是 FastAPI 封装batch_run.py是批量任务脚本。下面从核心代码开始写。6. 代码实现从单智能体到多智能体编排6.1 统一模型连接在agents/llm.py中封装模型的初始化逻辑避免每个文件重复加载import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def get_llm(temperature: float | None None): return ChatOpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), modelos.getenv(LLM_MODEL), temperaturefloat(temperature if temperature is not None else os.getenv(LLM_TEMPERATURE, 0.7)), )这里用到的ChatOpenAI可以兼容大部分 OpenAI 风格接口服务。要注意api_key和base_url的参数名不同langchain-openai版本可能略有差异如果运行时报参数错误按提示调整即可。6.2 定义多智能体角色接着定义每个智能体。这里的关键是“每个智能体只做一件事”并且把 Prompt 写得足够具体这样模型输出才稳定。先写婚礼策划师节点。它负责解析用户需求整理出预算、人数、日期、城市、风格等字段并给出策划方向。# agents/planner.py from langchain_core.prompts import ChatPromptTemplate PLANNER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一名资深婚礼策划师负责理解新人需求并制定整体策划方向。), (human, 请根据以下需求提取关键信息并输出策划方向。 需求{requirement} 请以 JSON 格式输出包含以下字段 - budget: 总预算 - guest_count: 宾客人数 - wedding_date: 婚礼日期 - city: 城市 - style: 婚礼风格 - theme: 推荐主题 - key_tasks: 需要重点关注的策划事项列表 ), ]) def planner_node(state): llm state[llm] requirement state[requirement] chain PLANNER_PROMPT | llm result chain.invoke({requirement: requirement}) state[planner_output] result.content return state预算顾问节点读上面的计划给出费用分配建议# agents/budget.py from langchain_core.prompts import ChatPromptTemplate BUDGET_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一名婚礼预算顾问擅长把总预算合理分配到各个项目。), (human, 根据策划师的输出给出详细的预算分配方案。 策划输出{planner_output} 请输出 JSON 格式包含 - items: 每个预算项目的名称、金额、占比 - suggestions: 预算控制建议 ), ]) def budget_node(state): llm state[llm] planner_output state.get(planner_output, ) chain BUDGET_PROMPT | llm result chain.invoke({planner_output: planner_output}) state[budget_output] result.content return state类似的场地推荐官节点根据城市和预算生成场地筛选建议供应商协调员节点输出摄影、化妆、餐饮、主持等品类注意事项时间线管理员节点生成婚礼当天的时间安排。每个节点的结构一致区别只在于 Prompt 不同。这样设计的好处是你可以在不改变整体图结构的情况下单独调整某个角色的能力。6.3 用 LangGraph 组装多智能体流程在graph.py中把节点串起来# graph.py from langgraph.graph import StateGraph, START, END from typing import TypedDict, Optional from agents.planner import planner_node from agents.budget import budget_node from agents.venue import venue_node from agents.supplier import supplier_node from agents.timeline import timeline_node class WeddingState(TypedDict): requirement: str llm: object planner_output: Optional[str] budget_output: Optional[str] venue_output: Optional[str] supplier_output: Optional[str] timeline_output: Optional[str] final_output: Optional[str] def final_node(state): parts [ state.get(planner_output, ), state.get(budget_output, ), state.get(venue_output, ), state.get(supplier_output, ), state.get(timeline_output, ), ] combined \n\n.join([p for p in parts if p]) state[final_output] combined return state def build_graph(llm): graph StateGraph(WeddingState) graph.add_node(planner, planner_node) graph.add_node(budget, budget_node) graph.add_node(venue, venue_node) graph.add_node(supplier, supplier_node) graph.add_node(timeline, timeline_node) graph.add_node(final, final_node) graph.add_edge(START, planner) graph.add_edge(planner, budget) graph.add_edge(budget, venue) graph.add_edge(venue, supplier) graph.add_edge(supplier, timeline) graph.add_edge(timeline, final) graph.add_edge(final, END) return graph.compile()这里使用的是最简单的串行流程每个节点按顺序执行。实际项目中你可以改成并行执行比如预算顾问、场地推荐官、供应商协调员互不依赖可以并行调用提升响应速度。LangGraph 节点之间的依赖关系完全由边决定这是它比单纯 for 循环灵活的地方。在main.py中写一个简单入口先验证多智能体流程能不能跑通# main.py from graph import build_graph from agents.llm import get_llm if __name__ __main__: llm get_llm() app build_graph(llm) requirement 预算15万宾客200人明年5月在北京办草坪婚礼偏好自然简约风格需要统筹摄影、化妆、餐饮和现场布置。 result app.invoke({requirement: requirement, llm: llm}) print( 最终策划方案 ) print(result[final_output])运行python main.py如果一切正常终端会输出经过多个智能体协作生成的策划内容。这一步是整个项目的核心验证点能跑通说明多智能体编排没有问题跑不通则先排查 API Key、网络和依赖版本。7. 用 Streamlit 搭建可视化界面命令行能跑通之后接下来把多智能体流程接进 Streamlit。新建app.py实现一个聊天式交互界面。# app.py import streamlit as st from agents.llm import get_llm from graph import build_graph st.set_page_config(page_title多智能体婚礼策划师, page_icon:material/favorite:) st.title(多智能体婚礼策划师) if messages not in st.session_state: st.session_state.messages [ {role: assistant, content: 你好请告诉我你的婚礼需求包括预算、人数、日期、城市、风格偏好等信息。} ] if graph not in st.session_state: llm get_llm() st.session_state.graph build_graph(llm) for msg in st.session_state.messages: st.chat_message(msg[role]).write(msg[content]) if prompt : st.chat_input(输入你的婚礼需求): st.session_state.messages.append({role: user, content: prompt}) st.chat_message(user).write(prompt) with st.chat_message(assistant): with st.spinner(多个智能体正在协作策划请稍候...): try: result st.session_state.graph.invoke({ requirement: prompt, llm: st.session_state.graph.nodes }) answer result[final_output] st.markdown(answer) st.session_state.messages.append({role: assistant, content: answer}) except Exception as e: st.error(f执行失败{e})需要说明一下上面st.session_state.graph.invoke()里的llm字段main.py是直接传入llm对象在 Streamlit 场景中由于st.session_state存储的限制和节点函数签名不同更稳妥的做法是把llm放进WeddingState而不是在节点内部重新初始化。也就是说节点函数里优先从state[llm]读取模型对象避免每个节点重复读取环境变量和新建连接。启动界面streamlit run app.py启动后浏览器会打开http://localhost:8501。在输入框里输入需求比如“预算15万宾客200人明年5月在北京办草坪婚礼偏好自然简约风格”点击发送就能看到多个智能体协作生成的结果。右上角的菜单里可以查看运行日志方便排查问题。8. 功能测试与效果验证8.1 基础链路测试建议先用一组结构化的需求做首次测试。输入示例预算15万宾客200人明年5月在北京办草坪婚礼偏好自然简约风格需要统筹摄影、化妆、餐饮和现场布置。预期结果策划师正确提取预算、人数、日期、城市、风格。预算顾问按摄影、化妆、餐饮、场地等维度给出分配比例。场地推荐官给出草坪婚礼场地的筛选维度。供应商协调员输出各品类注意事项。时间线管理员生成当天流程。判断成功的标准是最终输出包含以上五个维度的内容且每个维度的信息与前面节点输出的数据一致。如果预算顾问生成的金额加起来明显超出总预算说明 Prompt 约束还不够强需要调整。8.2 边界场景测试再试几组偏边界的需求看系统是否稳定。测试用例 1信息缺失。输入“我想办婚礼”没有预算、人数、日期。好的表现是策划师能主动列出需要补充的信息而不是硬编一个方案。测试用例 2极端预算。输入“预算5万宾客500人要在五星级酒店办”模型可能会给出不合理的建议。这时候要检查预算顾问是否提出“预算与需求不匹配”的风险提示。测试用例 3长文本输入。把所有需求写成一大段描述文字包含多个约束条件测试多智能体是否能逐条解析。这三个用例能帮你发现 Prompt 设计上的问题。在正式使用前至少把基础链路和第一个边界用例跑通。8.3 失败排查顺序如果测试过程中出现异常按下面的顺序排查先看 Streamlit 或终端里的报错信息。确认.env中的 API Key、Base URL、模型名是否填对。用curl或一段简单的 Python 代码直接调用模型 API排除模型服务本身的问题。确认langgraph的状态字段名是否与节点函数里读取的字段名一致。如果输出是 JSON 但解析失败检查 Prompt 是否要求模型只输出 JSON或用输出解析器处理。9. 接口 API 与批量任务扩展命令行和 Web 界面都验证通过后可以把它封装成 HTTP 接口供其他系统调用。新建api_service.py# api_service.py from fastapi import FastAPI from pydantic import BaseModel from agents.llm import get_llm from graph import build_graph app FastAPI() class WeddingRequest(BaseModel): requirement: str app.post(/generate_plan) def generate_plan(request: WeddingRequest): llm get_llm() graph build_graph(llm) result graph.invoke({requirement: request.requirement, llm: llm}) return {plan: result[final_output]}启动 API 服务uvicorn api_service:app --host 0.0.0.0 --port 8000这里要注意--host 0.0.0.0表示监听所有网卡地址如果只在本地调试建议改成127.0.0.1避免服务暴露到外部网络。用 curl 测试接口curl -X POST http://127.0.0.1:8000/generate_plan \ -H Content-Type: application/json \ -d {requirement: 预算10万宾客150人明年10月在上海办室内婚礼偏好复古风格}用 Python 请求接口import requests url http://127.0.0.1:8000/generate_plan payload { requirement: 预算10万宾客150人明年10月在上海办室内婚礼偏好复古风格 } response requests.post(url, jsonpayload, timeout120) print(response.json()[plan])接口跑通之后可以做批量任务。批量场景通常是把一批新人的需求存放在 CSV 里逐条调用多智能体流程把生成的方案保存成独立的 Markdown 文件。新建batch_run.py# batch_run.py import csv import os import time from agents.llm import get_llm from graph import build_graph def run_batch(csv_path: str, output_dir: str outputs): llm get_llm() graph build_graph(llm) os.makedirs(output_dir, exist_okTrue) with open(csv_path, r, encodingutf-8-sig) as f: reader csv.DictReader(f) for idx, row in enumerate(reader, start1): requirement f预算{row[budget]}宾客{row[guest_count]}人 \ f日期{row[date]}城市{row[city]}风格{row[style]} try: result graph.invoke({requirement: requirement, llm: llm}) output_path os.path.join(output_dir, fplan_{idx}.md) with open(output_path, w, encodingutf-8) as fout: fout.write(result[final_output]) print(f[{idx}] success - {output_path}) except Exception as e: print(f[{idx}] failed: {e}) time.sleep(1) if __name__ __main__: run_batch(data/requirements.csv)对应的data/requirements.csv示例budget,guest_count,date,city,style 150000,200,2025-05-18,北京,草坪婚礼 100000,150,2025-10-02,上海,复古风 80000,100,2025-06-07,杭州,新中式批量任务要加失败重试和日志。上面的脚本只是最简单的实现生产环境建议记录每个任务的执行耗时、token 消耗、失败原因并支持断点续跑避免中间某条出错后全部重来。10. 性能观察与成本控制这个项目调用云端大模型 API体感上的“性能瓶颈”主要在网络延迟和模型响应速度而不是本机 CPU 或显卡。单个节点一次调用可能耗时几秒到几十秒五个节点串行执行整体耗时可能达到几十秒。如果想提升速度可以把互不依赖的节点改成并行执行LangGraph 支持通过添加边的方式实现简单的并行分支具体写法可以参考官方文档。本地资源占用方面如果不下载本地模型Streamlit 和 FastAPI 的 CPU、内存占用都很低。真正的资源消耗在模型服务端体现为每次调用的 token 数和请求延迟。要观察每次节点调用的耗时可以在节点函数里加时间戳统计或者接 LangSmith 之类的追踪工具。成本控制可以从几个角度入手。第一选合适的模型不必所有节点都用最强模型像时间线整理这种结构化任务用小一点的模型可能够用。第二限制上下文长度每个节点的输入只传必要字段不要把上一个节点的全部原文都塞进去。第三做缓存如果多对新人的需求相似可以把典型方案的生成结果缓存起来。第四加频率限制批量任务里每两次请求之间加短暂延时既避免触发平台限流也方便控制预算。显存方面纯 API 模式基本不占显存。如果你之后换成 Ollama 跑本地模型显存占用取决于模型参数量、上下文长度和并发数需要根据自己的显卡单独测试。这个项目本身的设计目标就是让大家在普通电脑上也能体验多智能体开发不建议一开始就上 7B、13B 以上的本地模型。11. 常见问题与排查方法问题现象可能原因排查方式解决方案安装依赖失败Python 版本过低或网络问题python --version检查 pip 源升级 Python使用镜像源安装启动 main.py 报ModuleNotFoundError缺少依赖或导入路径错误在项目根目录运行命令检查 requirementspip install -r requirements.txt确认目录结构API 返回 401 或认证失败API Key 错误或余额不足用 curl 直接测试模型服务检查.env配置确认平台余额API 返回超时网络慢或模型服务负载高单独 curl 模型接口测试增加请求超时时间稍后重试智能体输出不是合法 JSONPrompt 约束不够打印节点原始返回内容在 Prompt 中强调只输出 JSON或用解析器兜底生成的预算分配不合理Prompt 缺少比例限制查看 budget 节点输出在 Prompt 中增加预算分配规则比如场地不超过 40%Streamlit 页面打不开端口被占用看终端日志检查 8501 端口streamlit run app.py --server.port 8502批量任务中途报错单条数据触发模型异常查看失败行日志加 try/except记录失败原因断点续跑API 服务只能在本地访问host 绑定问题检查 uvicorn host 参数本地调试用 127.0.0.1不要随意绑定 0.0.0.0最常见的问题还是环境问题。很多新手在没激活虚拟环境的情况下运行 pip结果装到了全局 Python项目里仍然找不到模块。每次打开新终端先确认命令行前面有(venv)标记再继续操作。12. 最佳实践与使用建议第一第一次跑通之前不要急着加功能。先按main.py的最小链路验证 API Key、模型连接和多智能体编排确认每个节点都能正常输出再去接 Streamlit 和 API 服务。第二Prompt 设计要具体。这个项目的核心质量取决于每个角色 Prompt 的约束力度建议给每个节点都定义清晰的“输入字段、输出字段、输出格式”甚至写几个 few-shot 示例。第三模型输出要兜底。不要把大模型返回的字符串直接当作结构化数据用。如果后续要做前端表格展示建议用 Pydantic 定义输出模型或让模型额外输出一个 JSON 版本解析失败时给出提示而不是整个页面报错。第四状态管理要统一。LangGraph 的 State 是整个流程的数据总线字段名一旦写错节点之间就会互相读不到数据建议把 State 的字段定义在独立文件里便于维护。第五接口服务要限制访问范围。FastAPI 启动时如果监听外网地址必须考虑鉴权问题至少加一个简单的 Token 校验避免被任意调用刷额度。第六涉及真实业务数据时务必完成脱敏和授权确认。用脱敏的测试数据开发用客户授权过的数据做验证不要把从公开渠道抓到的个人联系信息直接录入系统。第七批量任务要留“人工复核”出口。自动生成的方案只能作为初稿真实的婚礼策划还需要人工调整和确认。第八习惯性记录每次生成的参数和结果既能做好成本统计也能在模型表现波动时快速回退到更稳定的历史配置。13. 总结与下一步这个项目最值得尝试的点是让你用三个主流技术栈组合出一个“看得见、能操作、可扩展”的多智能体应用。LangChain 负责大模型调用链LangGraph 负责多智能体编排Streamlit 负责交互界面三者分工明确学习路径清晰。最优先验证的功能是main.py的多智能体链路只要这一步跑通后面的 Web 界面和 API 封装都是顺势而为。最容易踩的坑有两个一是环境依赖冲突建议全程使用虚拟环境二是 Prompt 对 JSON 输出的约束不够导致后续节点解析失败建议一开始就给每个智能体定义严格的输入输出格式。完成了基础链路后可以考虑从并行执行、记忆机制、知识库检索、更深度的合规校验这几个方向继续扩展。如果要把项目接到真实业务里请记住模型输出永远只能当草稿人工审核和合规确认不能省。
LangChain多智能体+Python+Streamlit：婚礼策划师应用实战
这次我们来看一个很有意思的实战型项目用 LangChain 多智能体架构加 Python 和 Streamlit做一个“婚礼策划师”应用。它不是概念演示而是把策划过程中的需求分析、预算规划、场地建议、时间线安排、供应商协调这些环节拆成多个智能体让每个智能体负责一个专业角色再由 LangGraph 统一编排最后通过 Streamlit 做出可视化操作界面。整个项目的核心是“多智能体怎么协作”而婚礼策划只是落地场景。这个项目最值得关注的有几点第一门槛不高默认走大模型 API 调用不要求本地有一张高显存显卡中等配置电脑完全能跑第二功能边界清晰每个智能体只做一件事方便调试和扩展后续换场景、加角色都很快第三有完整的 Web 界面不是只能在终端里敲命令适合做演示、接外包原型、或者给非技术用户试用第四可以继续封装成 API 服务也支持批量任务比如一次处理多对新人的策划需求。本文会带你把环境、架构、代码、界面、测试、API 封装、批量任务整个链路跑通。阅读本文需要一点 Python 基础至少知道怎么装依赖、怎么跑脚本。如果你正在学 LangChain 和 LangGraph又不想只看官方文档的抽象例子这个项目会是比较完整的练手素材。下面直接进入正文。1. 多智能体婚礼策划师核心能力速览能力项说明项目类型多智能体应用开发实战LangChain LangGraph Streamlit实现语言Python大模型接入默认使用 OpenAI 兼容接口可以填 DeepSeek、通义千问、Kimi 等平台的 API Key硬件要求调用云端 API 时基本不依赖本地 GPU中等配置电脑即可本地模型支持理论上可以换成 Ollama 等本地推理服务但显存和速度取决于模型大小需自行测试多智能体角色婚礼策划师、预算顾问、场地推荐官、供应商协调员、时间线管理员编排方式LangGraph 状态图各节点按流程依次执行并共享上下文Web 界面Streamlit 聊天式交互支持输入新人需求后自动生成完整方案API 服务可用 FastAPI 封装成 HTTP 接口供其他系统调用批量任务支持读取 CSV 需求列表批量生成策划方案 Markdown 文档适合场景LangChain 学习实战、多智能体入门、策划行业工具原型、企业内部流程演示需要说明一点这里说的“中配”不是指某个特定显卡型号而是指“不依赖本地大模型推理、只要能正常跑 Python 和浏览器”的日常开发环境。如果你要换成本地模型请以实际测试为准。2. 适用场景与使用边界这个项目适合三类人。第一类是正在学 LangChain 的开发者想搞明白 Prompt 模板、模型封装、链式调用、多智能体编排分别解决什么问题婚礼策划这个场景足够具体比“通用助手”更容易理解。第二类是产品原型开发者需要快速搭建一个“多角色协作 可视化界面”的 Demo用来给客户或领导演示。第三类是策划行业的技术爱好者想验证 AI 在方案生成、预算测算、时间线管理上能省多少人工。它不适合什么场景如果你想要的是一个生产级别的智能策划系统那还差得远。大模型存在幻觉生成的场地价格、供应商名单必须人工核实不同地区的婚礼习俗差异很大通用 Prompt 覆盖不了全部情况涉及真实客户姓名、联系方式、预算金额这类信息时直接丢给第三方 API 会有隐私风险。另外大模型 API 服务偶尔会不稳定如果做成线上系统必须有超时重试、降级方案和人工审核环节。合规方面要特别注意不要拿真实客户隐私数据做测试不要用模型生成的报价单直接发给客户不要把未经授权的真实商家信息当作“推荐结果”输出如果以后接入了图片生成、人脸相关的能力还需要确认肖像权和内容授权。技术本身是中性的但使用边界要清楚。3. 整体架构为什么用 LangGraph 编排多智能体多智能体项目最常见的误区是把多个 Prompt 放在一个数组里轮询假装它们是“多个智能体”。真正可维护的多智能体系统至少要满足三点每个智能体只负责一个职能智能体共享会话状态流程可以被控制、被观察、被修改。LangGraph 的状态图正好对应这套设计。本项目的流程是这样设计的用户输入婚礼需求包括预算、人数、日期、风格偏好、所在城市等随后第一个节点是“策划师”负责理解需求并拆解出待办事项之后“预算顾问”根据预算和人数输出分配建议“场地推荐官”结合城市和风格给出场地类型和筛选方向“供应商协调员”生成摄影、化妆、餐饮、主持等品类的注意事项“时间线管理员”负责把以上内容排成婚礼当天的日程。所有节点运行完后由一个汇总节点生成完整的策划方案 Markdown。LangGraph 在这里的价值是它让每个节点之间的“数据流”变得明确。你可以把状态定义为一个字典每个 Agent 从状态里读自己需要的字段再把产出写回去。后续加一个“应急预案智能体”或“宾客管理智能体”只需要新增节点和边不用推翻重写。从工程角度讲这种架构还方便调试。节点执行到哪一步、哪一步耗时最长、哪一步返回了非法 JSON都能通过日志定位。对学习者来说理解这个流程之后再去看 LangChain 官方文档或 LangGraph 示例会轻松很多。后面我们直接用代码把它搭出来。4. 环境准备与前置条件开始之前先确认本机环境满足下面这些条件。检查项建议值说明操作系统Windows 10/11、macOS、Linux 均可本项目的核心代码跨平台Python3.9 及以上建议 3.10 或 3.11LangChain 和 LangGraph 对高版本支持更好pip最新版本避免依赖解析问题API Key一个 OpenAI 兼容的模型服务 Key以 DeepSeek 或通义千问为例按平台要求开通网络能正常访问模型 API 服务具体域名以你选择的平台为准浏览器Chrome / Edge 等现代浏览器用于打开 Streamlit 界面磁盘预留 2GB 以上主要是 Python 依赖和日志模型文件不下载则占用很小如果你的电脑已经装了 Anaconda 或 Miniconda可以用虚拟环境隔离项目。如果没有直接用 Python 自带的 venv 也可以。这一步很重要因为 LangChain、Streamlit、FastAPI 都有各自的依赖混装在全局环境里容易冲突。检查 Python 版本python --version如果版本低于 3.9建议先升级 Python 环境。这里给出一套通用的虚拟环境创建命令实际路径按你自己的项目目录调整mkdir langchain-wedding-planner cd langchain-wedding-planner python -m venv venvWindows 下激活虚拟环境venv\Scripts\activatemacOS / Linux 下激活虚拟环境source venv/bin/activate激活后命令行前面会出现(venv)标记。接下来安装依赖。5. 安装部署与项目初始化在项目根目录下新建requirements.txt写入以下依赖langchain0.3.0 langchain-openai0.2.0 langgraph0.2.0 streamlit1.36.0 python-dotenv1.0.0 fastapi0.111.0 uvicorn0.30.0 pandas2.0.0安装命令pip install -r requirements.txt如果你的网络环境使用镜像源可以临时指定国内 PyPI 镜像例如pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple然后创建.env文件保存模型服务的配置。这里以 OpenAI 兼容接口为例LLM_API_KEYsk-你的密钥 LLM_BASE_URLhttps://api.deepseek.com/v1 LLM_MODELdeepseek-chat LLM_TEMPERATURE0.7如果使用其他平台把LLM_BASE_URL和LLM_MODEL换成你实际的服务地址和模型名。这里只写通用模板具体值以你的模型服务商提供的信息为准。项目目录结构建议这样组织langchain-wedding-planner/ ├── .env ├── requirements.txt ├── agents/ │ ├── __init__.py │ ├── llm.py │ ├── planner.py │ ├── budget.py │ ├── venue.py │ ├── supplier.py │ └── timeline.py ├── graph.py ├── app.py ├── api_service.py ├── batch_run.py ├── outputs/ └── data/ └── requirements.csvagents目录用来存放每个智能体的 Prompt 和节点逻辑graph.py负责用 LangGraph 组装整个多智能体流程app.py是 Streamlit 前端api_service.py是 FastAPI 封装batch_run.py是批量任务脚本。下面从核心代码开始写。6. 代码实现从单智能体到多智能体编排6.1 统一模型连接在agents/llm.py中封装模型的初始化逻辑避免每个文件重复加载import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def get_llm(temperature: float | None None): return ChatOpenAI( api_keyos.getenv(LLM_API_KEY), base_urlos.getenv(LLM_BASE_URL), modelos.getenv(LLM_MODEL), temperaturefloat(temperature if temperature is not None else os.getenv(LLM_TEMPERATURE, 0.7)), )这里用到的ChatOpenAI可以兼容大部分 OpenAI 风格接口服务。要注意api_key和base_url的参数名不同langchain-openai版本可能略有差异如果运行时报参数错误按提示调整即可。6.2 定义多智能体角色接着定义每个智能体。这里的关键是“每个智能体只做一件事”并且把 Prompt 写得足够具体这样模型输出才稳定。先写婚礼策划师节点。它负责解析用户需求整理出预算、人数、日期、城市、风格等字段并给出策划方向。# agents/planner.py from langchain_core.prompts import ChatPromptTemplate PLANNER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一名资深婚礼策划师负责理解新人需求并制定整体策划方向。), (human, 请根据以下需求提取关键信息并输出策划方向。 需求{requirement} 请以 JSON 格式输出包含以下字段 - budget: 总预算 - guest_count: 宾客人数 - wedding_date: 婚礼日期 - city: 城市 - style: 婚礼风格 - theme: 推荐主题 - key_tasks: 需要重点关注的策划事项列表 ), ]) def planner_node(state): llm state[llm] requirement state[requirement] chain PLANNER_PROMPT | llm result chain.invoke({requirement: requirement}) state[planner_output] result.content return state预算顾问节点读上面的计划给出费用分配建议# agents/budget.py from langchain_core.prompts import ChatPromptTemplate BUDGET_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一名婚礼预算顾问擅长把总预算合理分配到各个项目。), (human, 根据策划师的输出给出详细的预算分配方案。 策划输出{planner_output} 请输出 JSON 格式包含 - items: 每个预算项目的名称、金额、占比 - suggestions: 预算控制建议 ), ]) def budget_node(state): llm state[llm] planner_output state.get(planner_output, ) chain BUDGET_PROMPT | llm result chain.invoke({planner_output: planner_output}) state[budget_output] result.content return state类似的场地推荐官节点根据城市和预算生成场地筛选建议供应商协调员节点输出摄影、化妆、餐饮、主持等品类注意事项时间线管理员节点生成婚礼当天的时间安排。每个节点的结构一致区别只在于 Prompt 不同。这样设计的好处是你可以在不改变整体图结构的情况下单独调整某个角色的能力。6.3 用 LangGraph 组装多智能体流程在graph.py中把节点串起来# graph.py from langgraph.graph import StateGraph, START, END from typing import TypedDict, Optional from agents.planner import planner_node from agents.budget import budget_node from agents.venue import venue_node from agents.supplier import supplier_node from agents.timeline import timeline_node class WeddingState(TypedDict): requirement: str llm: object planner_output: Optional[str] budget_output: Optional[str] venue_output: Optional[str] supplier_output: Optional[str] timeline_output: Optional[str] final_output: Optional[str] def final_node(state): parts [ state.get(planner_output, ), state.get(budget_output, ), state.get(venue_output, ), state.get(supplier_output, ), state.get(timeline_output, ), ] combined \n\n.join([p for p in parts if p]) state[final_output] combined return state def build_graph(llm): graph StateGraph(WeddingState) graph.add_node(planner, planner_node) graph.add_node(budget, budget_node) graph.add_node(venue, venue_node) graph.add_node(supplier, supplier_node) graph.add_node(timeline, timeline_node) graph.add_node(final, final_node) graph.add_edge(START, planner) graph.add_edge(planner, budget) graph.add_edge(budget, venue) graph.add_edge(venue, supplier) graph.add_edge(supplier, timeline) graph.add_edge(timeline, final) graph.add_edge(final, END) return graph.compile()这里使用的是最简单的串行流程每个节点按顺序执行。实际项目中你可以改成并行执行比如预算顾问、场地推荐官、供应商协调员互不依赖可以并行调用提升响应速度。LangGraph 节点之间的依赖关系完全由边决定这是它比单纯 for 循环灵活的地方。在main.py中写一个简单入口先验证多智能体流程能不能跑通# main.py from graph import build_graph from agents.llm import get_llm if __name__ __main__: llm get_llm() app build_graph(llm) requirement 预算15万宾客200人明年5月在北京办草坪婚礼偏好自然简约风格需要统筹摄影、化妆、餐饮和现场布置。 result app.invoke({requirement: requirement, llm: llm}) print( 最终策划方案 ) print(result[final_output])运行python main.py如果一切正常终端会输出经过多个智能体协作生成的策划内容。这一步是整个项目的核心验证点能跑通说明多智能体编排没有问题跑不通则先排查 API Key、网络和依赖版本。7. 用 Streamlit 搭建可视化界面命令行能跑通之后接下来把多智能体流程接进 Streamlit。新建app.py实现一个聊天式交互界面。# app.py import streamlit as st from agents.llm import get_llm from graph import build_graph st.set_page_config(page_title多智能体婚礼策划师, page_icon:material/favorite:) st.title(多智能体婚礼策划师) if messages not in st.session_state: st.session_state.messages [ {role: assistant, content: 你好请告诉我你的婚礼需求包括预算、人数、日期、城市、风格偏好等信息。} ] if graph not in st.session_state: llm get_llm() st.session_state.graph build_graph(llm) for msg in st.session_state.messages: st.chat_message(msg[role]).write(msg[content]) if prompt : st.chat_input(输入你的婚礼需求): st.session_state.messages.append({role: user, content: prompt}) st.chat_message(user).write(prompt) with st.chat_message(assistant): with st.spinner(多个智能体正在协作策划请稍候...): try: result st.session_state.graph.invoke({ requirement: prompt, llm: st.session_state.graph.nodes }) answer result[final_output] st.markdown(answer) st.session_state.messages.append({role: assistant, content: answer}) except Exception as e: st.error(f执行失败{e})需要说明一下上面st.session_state.graph.invoke()里的llm字段main.py是直接传入llm对象在 Streamlit 场景中由于st.session_state存储的限制和节点函数签名不同更稳妥的做法是把llm放进WeddingState而不是在节点内部重新初始化。也就是说节点函数里优先从state[llm]读取模型对象避免每个节点重复读取环境变量和新建连接。启动界面streamlit run app.py启动后浏览器会打开http://localhost:8501。在输入框里输入需求比如“预算15万宾客200人明年5月在北京办草坪婚礼偏好自然简约风格”点击发送就能看到多个智能体协作生成的结果。右上角的菜单里可以查看运行日志方便排查问题。8. 功能测试与效果验证8.1 基础链路测试建议先用一组结构化的需求做首次测试。输入示例预算15万宾客200人明年5月在北京办草坪婚礼偏好自然简约风格需要统筹摄影、化妆、餐饮和现场布置。预期结果策划师正确提取预算、人数、日期、城市、风格。预算顾问按摄影、化妆、餐饮、场地等维度给出分配比例。场地推荐官给出草坪婚礼场地的筛选维度。供应商协调员输出各品类注意事项。时间线管理员生成当天流程。判断成功的标准是最终输出包含以上五个维度的内容且每个维度的信息与前面节点输出的数据一致。如果预算顾问生成的金额加起来明显超出总预算说明 Prompt 约束还不够强需要调整。8.2 边界场景测试再试几组偏边界的需求看系统是否稳定。测试用例 1信息缺失。输入“我想办婚礼”没有预算、人数、日期。好的表现是策划师能主动列出需要补充的信息而不是硬编一个方案。测试用例 2极端预算。输入“预算5万宾客500人要在五星级酒店办”模型可能会给出不合理的建议。这时候要检查预算顾问是否提出“预算与需求不匹配”的风险提示。测试用例 3长文本输入。把所有需求写成一大段描述文字包含多个约束条件测试多智能体是否能逐条解析。这三个用例能帮你发现 Prompt 设计上的问题。在正式使用前至少把基础链路和第一个边界用例跑通。8.3 失败排查顺序如果测试过程中出现异常按下面的顺序排查先看 Streamlit 或终端里的报错信息。确认.env中的 API Key、Base URL、模型名是否填对。用curl或一段简单的 Python 代码直接调用模型 API排除模型服务本身的问题。确认langgraph的状态字段名是否与节点函数里读取的字段名一致。如果输出是 JSON 但解析失败检查 Prompt 是否要求模型只输出 JSON或用输出解析器处理。9. 接口 API 与批量任务扩展命令行和 Web 界面都验证通过后可以把它封装成 HTTP 接口供其他系统调用。新建api_service.py# api_service.py from fastapi import FastAPI from pydantic import BaseModel from agents.llm import get_llm from graph import build_graph app FastAPI() class WeddingRequest(BaseModel): requirement: str app.post(/generate_plan) def generate_plan(request: WeddingRequest): llm get_llm() graph build_graph(llm) result graph.invoke({requirement: request.requirement, llm: llm}) return {plan: result[final_output]}启动 API 服务uvicorn api_service:app --host 0.0.0.0 --port 8000这里要注意--host 0.0.0.0表示监听所有网卡地址如果只在本地调试建议改成127.0.0.1避免服务暴露到外部网络。用 curl 测试接口curl -X POST http://127.0.0.1:8000/generate_plan \ -H Content-Type: application/json \ -d {requirement: 预算10万宾客150人明年10月在上海办室内婚礼偏好复古风格}用 Python 请求接口import requests url http://127.0.0.1:8000/generate_plan payload { requirement: 预算10万宾客150人明年10月在上海办室内婚礼偏好复古风格 } response requests.post(url, jsonpayload, timeout120) print(response.json()[plan])接口跑通之后可以做批量任务。批量场景通常是把一批新人的需求存放在 CSV 里逐条调用多智能体流程把生成的方案保存成独立的 Markdown 文件。新建batch_run.py# batch_run.py import csv import os import time from agents.llm import get_llm from graph import build_graph def run_batch(csv_path: str, output_dir: str outputs): llm get_llm() graph build_graph(llm) os.makedirs(output_dir, exist_okTrue) with open(csv_path, r, encodingutf-8-sig) as f: reader csv.DictReader(f) for idx, row in enumerate(reader, start1): requirement f预算{row[budget]}宾客{row[guest_count]}人 \ f日期{row[date]}城市{row[city]}风格{row[style]} try: result graph.invoke({requirement: requirement, llm: llm}) output_path os.path.join(output_dir, fplan_{idx}.md) with open(output_path, w, encodingutf-8) as fout: fout.write(result[final_output]) print(f[{idx}] success - {output_path}) except Exception as e: print(f[{idx}] failed: {e}) time.sleep(1) if __name__ __main__: run_batch(data/requirements.csv)对应的data/requirements.csv示例budget,guest_count,date,city,style 150000,200,2025-05-18,北京,草坪婚礼 100000,150,2025-10-02,上海,复古风 80000,100,2025-06-07,杭州,新中式批量任务要加失败重试和日志。上面的脚本只是最简单的实现生产环境建议记录每个任务的执行耗时、token 消耗、失败原因并支持断点续跑避免中间某条出错后全部重来。10. 性能观察与成本控制这个项目调用云端大模型 API体感上的“性能瓶颈”主要在网络延迟和模型响应速度而不是本机 CPU 或显卡。单个节点一次调用可能耗时几秒到几十秒五个节点串行执行整体耗时可能达到几十秒。如果想提升速度可以把互不依赖的节点改成并行执行LangGraph 支持通过添加边的方式实现简单的并行分支具体写法可以参考官方文档。本地资源占用方面如果不下载本地模型Streamlit 和 FastAPI 的 CPU、内存占用都很低。真正的资源消耗在模型服务端体现为每次调用的 token 数和请求延迟。要观察每次节点调用的耗时可以在节点函数里加时间戳统计或者接 LangSmith 之类的追踪工具。成本控制可以从几个角度入手。第一选合适的模型不必所有节点都用最强模型像时间线整理这种结构化任务用小一点的模型可能够用。第二限制上下文长度每个节点的输入只传必要字段不要把上一个节点的全部原文都塞进去。第三做缓存如果多对新人的需求相似可以把典型方案的生成结果缓存起来。第四加频率限制批量任务里每两次请求之间加短暂延时既避免触发平台限流也方便控制预算。显存方面纯 API 模式基本不占显存。如果你之后换成 Ollama 跑本地模型显存占用取决于模型参数量、上下文长度和并发数需要根据自己的显卡单独测试。这个项目本身的设计目标就是让大家在普通电脑上也能体验多智能体开发不建议一开始就上 7B、13B 以上的本地模型。11. 常见问题与排查方法问题现象可能原因排查方式解决方案安装依赖失败Python 版本过低或网络问题python --version检查 pip 源升级 Python使用镜像源安装启动 main.py 报ModuleNotFoundError缺少依赖或导入路径错误在项目根目录运行命令检查 requirementspip install -r requirements.txt确认目录结构API 返回 401 或认证失败API Key 错误或余额不足用 curl 直接测试模型服务检查.env配置确认平台余额API 返回超时网络慢或模型服务负载高单独 curl 模型接口测试增加请求超时时间稍后重试智能体输出不是合法 JSONPrompt 约束不够打印节点原始返回内容在 Prompt 中强调只输出 JSON或用解析器兜底生成的预算分配不合理Prompt 缺少比例限制查看 budget 节点输出在 Prompt 中增加预算分配规则比如场地不超过 40%Streamlit 页面打不开端口被占用看终端日志检查 8501 端口streamlit run app.py --server.port 8502批量任务中途报错单条数据触发模型异常查看失败行日志加 try/except记录失败原因断点续跑API 服务只能在本地访问host 绑定问题检查 uvicorn host 参数本地调试用 127.0.0.1不要随意绑定 0.0.0.0最常见的问题还是环境问题。很多新手在没激活虚拟环境的情况下运行 pip结果装到了全局 Python项目里仍然找不到模块。每次打开新终端先确认命令行前面有(venv)标记再继续操作。12. 最佳实践与使用建议第一第一次跑通之前不要急着加功能。先按main.py的最小链路验证 API Key、模型连接和多智能体编排确认每个节点都能正常输出再去接 Streamlit 和 API 服务。第二Prompt 设计要具体。这个项目的核心质量取决于每个角色 Prompt 的约束力度建议给每个节点都定义清晰的“输入字段、输出字段、输出格式”甚至写几个 few-shot 示例。第三模型输出要兜底。不要把大模型返回的字符串直接当作结构化数据用。如果后续要做前端表格展示建议用 Pydantic 定义输出模型或让模型额外输出一个 JSON 版本解析失败时给出提示而不是整个页面报错。第四状态管理要统一。LangGraph 的 State 是整个流程的数据总线字段名一旦写错节点之间就会互相读不到数据建议把 State 的字段定义在独立文件里便于维护。第五接口服务要限制访问范围。FastAPI 启动时如果监听外网地址必须考虑鉴权问题至少加一个简单的 Token 校验避免被任意调用刷额度。第六涉及真实业务数据时务必完成脱敏和授权确认。用脱敏的测试数据开发用客户授权过的数据做验证不要把从公开渠道抓到的个人联系信息直接录入系统。第七批量任务要留“人工复核”出口。自动生成的方案只能作为初稿真实的婚礼策划还需要人工调整和确认。第八习惯性记录每次生成的参数和结果既能做好成本统计也能在模型表现波动时快速回退到更稳定的历史配置。13. 总结与下一步这个项目最值得尝试的点是让你用三个主流技术栈组合出一个“看得见、能操作、可扩展”的多智能体应用。LangChain 负责大模型调用链LangGraph 负责多智能体编排Streamlit 负责交互界面三者分工明确学习路径清晰。最优先验证的功能是main.py的多智能体链路只要这一步跑通后面的 Web 界面和 API 封装都是顺势而为。最容易踩的坑有两个一是环境依赖冲突建议全程使用虚拟环境二是 Prompt 对 JSON 输出的约束不够导致后续节点解析失败建议一开始就给每个智能体定义严格的输入输出格式。完成了基础链路后可以考虑从并行执行、记忆机制、知识库检索、更深度的合规校验这几个方向继续扩展。如果要把项目接到真实业务里请记住模型输出永远只能当草稿人工审核和合规确认不能省。
