专栏：多智能体与大模型实战

源码获取：awesome-llm-apps/advanced_ai_agents/multi_agent_apps/agent_teams/multimodal_design_agent_team at main · Shubhamsaboo/awesome-llm-apps

文章导读：随着大模型Agent技术普及，Multi-Agent（多智能体）成为AI落地核心方向。本文带来一套工业级高质量多智能体项目，全程拆解代码逻辑、技术架构与设计模式，手把手带你吃透实战开发，新手也能快速上手，适合学习、毕设、项目改造直接复用！

---

📌 项目概述

这是一套模拟真实企业团队的AI多智能体项目咨询系统，基于Python开发，无需复杂前端配置，通过Streamlit快速搭建可视化界面，依托Agency-Swarm框架实现5大AI角色分工协作，自动完成项目可行性分析、技术方案输出、产品规划、开发落地、客户全流程管理。

核心功能

• 可视化表单输入项目基础信息，操作零门槛

• 5大AI智能体各司其职，模拟真实职场协作流程

• 标准化工具调用，保证输出结果规范可控

• 多标签页分类展示结果，逻辑清晰一目了然

• 会话状态持久化，保留历史分析记录

适用场景

• 项目可行性评估与需求分析

• AI技术咨询与架构设计助手

• 多智能体协作学习标杆项目

• 毕设/课设AI类项目快速改造

---

🛠️ 核心技术栈

技术/框架	用途
Python	核心开发语言，全流程实现
Streamlit	快速构建Web可视化界面，无需前端功底
Agency-Swarm	多智能体协作框架，实现Agent调度与通信
OpenAI API	大模型支撑，赋予AI智能体决策能力
Pydantic	工具参数校验，规范AI输入输出格式

---

🏗️ 项目架构与设计模式

本项目严格遵循Multi-Agent五大经典设计模式，架构低耦合、高可用，完全符合工业级开发规范，也是面试、毕设的加分亮点：

1. 分层协作模式：CEO统筹全局，其他Agent执行细分任务

2. 工具调用模式：自定义标准化工具，避免AI自由输出失控

3. 责任链模式：按固定流程分步执行，任务传递有序不混乱

4. 共享状态模式：多Agent共用上下文数据，避免重复计算

5. 中介者模式：Agency统一调度所有Agent，降低模块耦合

---

💻 完整代码逐模块深度解读

模块一：依赖库导入

导入项目所需核心库，包含类型注解、多智能体框架、参数校验、Web界面相关依赖，做好基础铺垫。

from typing import List, Literal, Dict, Optional
from agency_swarm import Agent, Agency, set_openai_key, BaseTool
from pydantic import Field, BaseModel
import streamlit as st

模块二：自定义智能体工具（核心）

工具是Multi-Agent系统的核心抓手，通过Pydantic定义规范参数，强制AI按固定格式执行任务，同时通过共享状态实现Agent间数据互通。

1. 项目需求分析工具

class AnalyzeProjectRequirements(BaseTool):
    project_name: str = Field(..., description="Name of the project")
    project_description: str = Field(..., description="Project description and goals")
    project_type: Literal["Web Application", "Mobile App", "API Development", 
                         "Data Analytics", "AI/ML Solution", "Other"] = Field(..., 
                         description="Type of project")
    budget_range: Literal["$10k-$25k", "$25k-$50k", "$50k-$100k", "$100k+"] = Field(..., 
                         description="Budget range for the project")

    class ToolConfig:
        name = "analyze_project"
        description = "Analyzes project requirements and feasibility"
        one_call_at_a_time = True

    def run(self) -> str:
        """Analyzes project and stores results in shared state"""
        # 防止重复分析，保证流程唯一性
        if self._shared_state.get("project_analysis", None) is not None:
            raise ValueError("Project analysis already exists. Please proceed with technical specification.")
        
        # 生成项目分析结果，存入共享状态
        analysis = {
            "name": self.project_name,
            "type": self.project_type,
            "complexity": "high",
            "timeline": "6 months",
            "budget_feasibility": "within range",
            "requirements": ["Scalable architecture", "Security", "API integration"]
        }
        
        self._shared_state.set("project_analysis", analysis)
        return "Project analysis completed. Please proceed with technical specification."

核心作用：专属CEO智能体使用，完成项目初步可行性评估，结果存入共享空间，供后续CTO、产品经理等Agent直接调用，避免重复请求用户数据。

2. 技术规格生成工具

class CreateTechnicalSpecification(BaseTool):
    architecture_type: Literal["monolithic", "microservices", "serverless", "hybrid"] = Field(
        ..., 
        description="Proposed architecture type"
    )
    core_technologies: str = Field(
        ..., 
        description="Comma-separated list of main technologies and frameworks"
    )
    scalability_requirements: Literal["high", "medium", "low"] = Field(
        ..., 
        description="Scalability needs"
    )

    class ToolConfig:
        name = "create_technical_spec"
        description = "Creates technical specifications based on project analysis"
        one_call_at_a_time = True

    def run(self) -> str:
        """Creates technical specification based on analysis"""
        # 校验前置任务是否完成，遵循责任链模式
        project_analysis = self._shared_state.get("project_analysis", None)
        if project_analysis is None:
            raise ValueError("Please analyze project requirements first using AnalyzeProjectRequirements tool.")
        
        # 生成技术方案，存入共享状态
        spec = {
            "project_name": project_analysis["name"],
            "architecture": self.architecture_type,
            "technologies": self.core_technologies.split(","),
            "scalability": self.scalability_requirements
        }
        
        self._shared_state.set("technical_specification", spec)
        return f"Technical specification created for {project_analysis['name']}."

核心作用：专属CTO智能体使用，基于CEO的分析结果生成标准化技术方案，包含架构选型、技术栈、扩展性要求，衔接后续开发任务。

模块三：会话状态初始化

Streamlit页面刷新会重置数据，通过session_state持久化API密钥和历史对话记录，提升用户使用体验。

def init_session_state() -> None:
    """Initialize session state variables"""
    if 'messages' not in st.session_state:
        st.session_state.messages = []
    if 'api_key' not in st.session_state:
        st.session_state.api_key = None

模块四：主函数与界面逻辑（前端+后端联动）

包含页面配置、侧边栏API密钥输入、项目信息表单、智能体创建、任务调度、结果展示全流程，是项目的入口核心。

def main() -> None:
    # 页面基础配置
    st.set_page_config(page_title="AI Services Agency", layout="wide")
    init_session_state()
    
    st.title("🚀 AI Services Agency")
    
    # 侧边栏：API密钥配置
    with st.sidebar:
        st.header("🔑 API Configuration")
        openai_api_key = st.text_input(
            "OpenAI API Key",
            type="password",
            help="Enter your OpenAI API key to continue"
        )

        if openai_api_key:
            st.session_state.api_key = openai_api_key
            st.success("API Key accepted!")
        else:
            st.warning("⚠️ Please enter your OpenAI API Key to proceed")
            st.markdown("[Get your API key here](https://platform.openai.com/api-keys)")
            return
        
    # 初始化OpenAI密钥
    set_openai_key(st.session_state.api_key)
    api_headers = {"Authorization": f"Bearer {st.session_state.api_key}"}
    
    # 项目信息输入表单
    with st.form("project_form"):
        st.subheader("Project Details")
        
        project_name = st.text_input("Project Name")
        project_description = st.text_area(
            "Project Description",
            help="Describe the project, its goals, and any specific requirements"
        )
        
        # 双列布局，优化界面展示
        col1, col2 = st.columns(2)
        with col1:
            project_type = st.selectbox(
                "Project Type",
                ["Web Application", "Mobile App", "API Development", 
                 "Data Analytics", "AI/ML Solution", "Other"]
            )
            timeline = st.selectbox(
                "Expected Timeline",
                ["1-2 months", "3-4 months", "5-6 months", "6+ months"]
            )
        
        with col2:
            budget_range = st.selectbox(
                "Budget Range",
                ["$10k-$25k", "$25k-$50k", "$50k-$100k", "$100k+"]
            )
            priority = st.selectbox(
                "Project Priority",
                ["High", "Medium", "Low"]
            )
        
        tech_requirements = st.text_area(
            "Technical Requirements (optional)",
            help="Any specific technical requirements or preferences"
        )
        
        special_considerations = st.text_area(
            "Special Considerations (optional)",
            help="Any additional information or special requirements"
        )
        
        submitted = st.form_submit_button("Analyze Project")
        
        # 表单提交后执行多智能体分析
        if submitted and project_name and project_description:
            try:
                # 创建5大AI智能体，分工明确
                ceo = Agent(
                    name="Project Director",
                    description="You are a CEO of multiple companies in the past and have a lot of experience in evaluating projects and making strategic decisions.",
                    instructions="""
                    You are an experienced CEO who evaluates projects. Follow these steps strictly:
                    1. FIRST, use the AnalyzeProjectRequirements tool with required parameters
                    2. WAIT for the analysis to complete before proceeding
                    3. Review the analysis results and provide strategic recommendations
                    """,
                    tools=[AnalyzeProjectRequirements],
                    api_headers=api_headers,
                    temperature=0.7,
                    max_prompt_tokens=25000
                )

                cto = Agent(
                    name="Technical Architect",
                    description="Senior technical architect with deep expertise in system design.",
                    instructions="""
                    1. WAIT for the project analysis to be completed by the CEO
                    2. Use the CreateTechnicalSpecification tool with appropriate parameters
                    3. Review the technical specification and provide additional recommendations
                    """,
                    tools=[CreateTechnicalSpecification],
                    api_headers=api_headers,
                    temperature=0.5,
                    max_prompt_tokens=25000
                )

                # 产品经理、首席开发、客户成功经理智能体
                product_manager = Agent(
                    name="Product Manager",
                    description="Experienced product manager focused on delivery excellence.",
                    instructions="Manage project scope, define product requirements and build roadmap",
                    api_headers=api_headers,
                    temperature=0.4
                )

                developer = Agent(
                    name="Lead Developer",
                    description="Senior developer with full-stack expertise.",
                    instructions="Plan technical implementation and provide effort estimates",
                    api_headers=api_headers,
                    temperature=0.3
                )

                client_manager = Agent(
                    name="Client Success Manager",
                    description="Experienced client manager focused on project delivery.",
                    instructions="Ensure client satisfaction and manage expectations",
                    api_headers=api_headers,
                    temperature=0.6
                )

                # 创建Agency，作为智能体调度中枢
                agency = Agency(
                    [
                        ceo, cto, product_manager, developer, client_manager,
                        [ceo, cto], [ceo, product_manager], [ceo, developer], [ceo, client_manager],
                        [cto, developer], [product_manager, developer], [product_manager, client_manager]
                    ],
                    async_mode='threading',
                    shared_files='shared_files'
                )
                
                # 封装项目信息
                project_info = {
                    "name": project_name,
                    "description": project_description,
                    "type": project_type,
                    "timeline": timeline,
                    "budget": budget_range,
                    "priority": priority,
                    "technical_requirements": tech_requirements,
                    "special_considerations": special_considerations
                }

                st.session_state.messages.append({"role": "user", "content": str(project_info)})
                
                # 执行多智能体分析，加载中提示
                with st.spinner("AI Services Agency is analyzing your project..."):
                    try:
                        # 依次调用各智能体，遵循责任链模式
                        ceo_response = agency.get_completion(
                            message=f"Analyze this project using the AnalyzeProjectRequirements tool: Project Name:{project_name}, Description:{project_description}, Type:{project_type}, Budget:{budget_range}",
                            recipient_agent=ceo
                        )
                        
                        cto_response = agency.get_completion(
                            message="Create technical specification based on project analysis",
                            recipient_agent=cto
                        )
                        
                        pm_response = agency.get_completion(
                            message=f"Analyze project management aspects: {str(project_info)}",
                            recipient_agent=product_manager
                        )

                        developer_response = agency.get_completion(
                            message=f"Analyze technical implementation: {str(project_info)}",
                            recipient_agent=developer
                        )
                        
                        client_response = agency.get_completion(
                            message=f"Analyze client success aspects: {str(project_info)}",
                            recipient_agent=client_manager
                        )
                        
                        # 多标签页分类展示结果
                        tabs = st.tabs([
                            "CEO's Project Analysis",
                            "CTO's Technical Specification",
                            "Product Manager's Plan",
                            "Developer's Implementation",
                            "Client Success Strategy"
                        ])
                        
                        with tabs[0]:
                            st.markdown("## CEO's Strategic Analysis")
                            st.markdown(ceo_response)
                            st.session_state.messages.append({"role": "assistant", "content": ceo_response})
                        
                        with tabs[1]:
                            st.markdown("## CTO's Technical Specification")
                            st.markdown(cto_response)
                        
                        with tabs[2]:
                            st.markdown("## Product Manager's Plan")
                            st.markdown(pm_response)
                        
                        with tabs[3]:
                            st.markdown("## Lead Developer's Development Plan")
                            st.markdown(developer_response)
                        
                        with tabs[4]:
                            st.markdown("## Client Success Strategy")
                            st.markdown(client_response)

                    except Exception as e:
                        st.error(f"Error during analysis: {str(e)}")
                        st.error("Please check your inputs and API key and try again.")

            except Exception as e:
                st.error(f"Error during analysis: {str(e)}")
                st.error("Please check your API key and try again.")

    # 侧边栏：历史记录管理
    with st.sidebar:
        st.subheader("Options")
        if st.checkbox("Show Analysis History"):
            for message in st.session_state.messages:
                with st.chat_message(message["role"]):
                    st.markdown(message["content"])
        
        if st.button("Clear History"):
            st.session_state.messages = []
            st.rerun()

if __name__ == "__main__":
    main()

---

🚀 项目运行方式

1. 安装依赖库

   pip install streamlit agency-swarm pydantic openai

2. 准备OpenAI API密钥（平台官网申请）

3. 保存代码为app.py，执行运行命令 streamlit run app.py

4. 浏览器打开生成的本地地址，输入API密钥，填写项目信息即可使用

---

✨ 项目亮点总结

1. 代码规范高质量：结构清晰，注释完善，遵循PEP8规范，可直接用于学习和项目改造

2. Multi-Agent设计标准：完美落地五大设计模式，吃透这份代码就能掌握多智能体核心逻辑

3. 开箱即用：纯Python开发，无需前端配置，新手也能快速运行

4. 扩展性极强：可新增AI智能体、自定义工具、优化界面样式

5. 实战价值拉满：贴合真实业务场景，毕设、求职、个人项目改造都适用

---

💬 文末结语

这套Multi-Agent项目是市面上少有的可直接运行、架构标准、逻辑完整的实战案例，没有冗余代码，没有复杂配置，非常适合作为多智能体入门的标杆项目。吃透本文内容，就能轻松应对大模型Agent相关的学习、面试和项目开发，后续还能在此基础上拓展更多功能，打造专属的AI智能团队。

觉得文章对你有帮助的小伙伴，欢迎点赞👍、收藏⭐、评论💬，关注我，后续持续分享更多大模型、多智能体实战干货！

---

相关话题：#Multi-Agent #多智能体 #Agency-Swarm #Streamlit #AI项目实战 #大模型Agent