章节5：结业项目实战

本章导学

本模块的结业项目分为两大实战项目，涵盖从需求分析、架构设计到编码实现、部署上线的完整流程。项目一为多Agent协作智能办公系统（AutoGen+LangChain+RAG+vLLM），项目二为企业级AI客服中台（FastAPI+RAG+Agent+Docker）。通过这两个项目，你将把本模块所学的全部知识点融会贯通，真正具备企业级AI Agent系统的开发能力。

学习目标

能够独立完成多Agent协作系统的需求分析与架构设计
掌握AutoGen+LangChain+RAG+vLLM的技术整合
掌握企业级AI客服中台的全栈开发
学会项目文档撰写和代码交付规范
具备AI Agent项目的演示和讲解能力

项目一：多Agent协作智能办公系统

1.1 项目概述

项目名称：Multi-Agent Smart Office System

技术栈：AutoGen + LangChain + RAG + vLLM + FastAPI + Streamlit

项目目标：构建一个多Agent协作的智能办公系统，能自动完成以下办公任务：

📧 邮件撰写与回复：根据上下文自动生成邮件
📊 数据分析报表：查询数据库生成分析报告
📝 文档起草与审核：协作完成文档编写和审查
🔍 信息检索与总结：从知识库中检索并总结信息

1.2 系统架构

┌──────────────────────────────────────────┐
│              Streamlit UI                 │  <- 用户界面
└────────────────┬─────────────────────────┘
                 │
┌────────────────┴─────────────────────────┐
│           FastAPI Gateway                 │  <- API网关
└────────────────┬─────────────────────────┘
                 │
┌────────────────┴─────────────────────────┐
│         AutoGen Multi-Agent Core         │  <- 多Agent核心
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐    │
│  │邮件Agent│ │数据Agent│ │文档Agent│ │搜索Agent│    │
│  └──────┘ └──────┘ └──────┘ └──────┘    │
└────────┬────────┬────────┬────────┬──────┘
         │        │        │        │
    ┌────┘   ┌────┘   ┌────┘   ┌────┘
    │        │        │        │
┌───┴──┐ ┌──┴───┐ ┌──┴───┐ ┌──┴───┐
│vLLM  │ │RAG   │ │DB    │ │API   │
│推理  │ │知识库 │ │查询  │ │集成  │
└──────┘ └──────┘ └──────┘ └──────┘

1.3 技术方案详解

1.3.1 Agent定义

python

# agents/work_agents.py
"""智能办公系统Agent定义"""

import autogen
from typing import Optional

# LLM配置
llm_config = {
    "config_list": [
        {
            "model": "Qwen/Qwen2.5-7B-Instruct",
            "base_url": "http://vllm-service:8000/v1",
            "api_key": "not-needed"
        }
    ],
    "temperature": 0.7,
    "max_tokens": 2048
}

# ========= 邮件Agent =========
email_agent = autogen.AssistantAgent(
    name="邮件助手",
    system_message="""你是一名专业的企业邮件助手。你擅长：
1. 根据上下文撰写正式、得体的商务邮件
2. 邮件分类（询问、回复、通知、纪要）
3. 提取邮件关键信息
4. 邮件回复建议回复

请始终使用以下格式输出邮件：
---
收件人：[收件人]
主题：[邮件主题]
正文：[邮件正文]
---

完成后请说"邮件已完成"。""",
    llm_config=llm_config
)

# ========= 数据分析Agent =========
data_agent = autogen.AssistantAgent(
    name="数据分析师",
    system_message="""你是一名企业数据分析师。你擅长：
1. 将自然语言查询转化为SQL
2. 分析数据趋势和异常
3. 生成数据报表和可视化建议
4. 提供数据驱动的业务洞察

分析完成后请说"分析已完成"。""",
    llm_config=llm_config
)

# ========= 文档撰写Agent =========
doc_agent = autogen.AssistantAgent(
    name="文档撰写员",
    system_message="""你是一名文档撰写专家。你擅长：
1. 撰写工作报告、技术文档、会议纪要
2. 优化文档结构和表达
3. 确保文档格式规范
4. 文档版本管理

文档完成后请说"文档已完成"。""",
    llm_config=llm_config
)

# ========= 文档审查Agent =========
reviewer_agent = autogen.AssistantAgent(
    name="文档审查员",
    system_message="""你是一名严格的质量审查员。审查时重点关注：
1. 内容的准确性和完整性
2. 逻辑连贯性
3. 格式规范性
4. 语言表达是否清晰

审查通过的标准：以上4点全部通过。
若发现问题，请明确指出并给出修改建议。
只有当所有问题都解决后才说"审查通过"。""",
    llm_config=llm_config
)

# ========= 信息检索Agent =========
search_agent = autogen.AssistantAgent(
    name="信息检索员",
    system_message="""你是一名信息检索专家。你擅长：
1. 从知识库中检索相关信息
2. 总结和提炼关键信息
3. 验证信息的准确性
4. 标注信息来源

检索完成请说"检索已完成"。""",
    llm_config=llm_config
)

1.3.2 RAG知识库（LangChain + ChromaDB）

python

# rag/knowledge_base.py
"""企业知识库模块"""

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.document_loaders import DirectoryLoader, TextLoader
import os

class EnterpriseKnowledgeBase:
    """企业知识库"""
    
    def __init__(self, persist_dir="./kb_store"):
        self.persist_dir = persist_dir
        self.embeddings = HuggingFaceEmbeddings(
            model_name="BAAI/bge-large-zh-v1.5",
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True}
        )
    
    def build(self, docs_dir: str):
        """构建知识库"""
        # 加载文档
        loader = DirectoryLoader(
            docs_dir,
            glob="**/*.{md,txt,pdf}",
            loader_cls=TextLoader,
            loader_kwargs={"encoding": "utf-8"}
        )
        documents = loader.load()
        
        # 分块
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", "。", "！", "？", "；"]
        )
        chunks = splitter.split_documents(documents)
        
        # 构建向量库
        vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        vectorstore.persist()
        return vectorstore
    
    def load(self):
        """加载已有知识库"""
        return Chroma(
            persist_directory=self.persist_dir,
            embedding_function=self.embeddings
        )
    
    def search(self, query: str, k: int = 5) -> list:
        """检索知识库"""
        vectorstore = self.load()
        results = vectorstore.similarity_search_with_score(query, k=k)
        return [
            {
                "content": doc.page_content,
                "source": doc.metadata.get("source", "未知"),
                "score": score
            }
            for doc, score in results
        ]

1.3.3 AutoGen工作流

python

# workflows/office_workflow.py
"""办公工作流"""

import autogen
from typing import Optional

class OfficeWorkflow:
    """智能办公工作流管理器"""
    
    def __init__(self, llm_config: dict):
        self.llm_config = llm_config
        self.agents = self._init_agents()
    
    def _init_agents(self):
        """初始化Agent"""
        return {
            "email": email_agent,
            "data": data_agent,
            "doc": doc_agent,
            "reviewer": reviewer_agent,
            "search": search_agent
        }
    
    def email_workflow(self, task: str) -> str:
        """邮件工作流：撰写→审查"""
        user_proxy = autogen.UserProxyAgent(
            name="用户",
            human_input_mode="NEVER",
            max_consecutive_auto_reply=10,
            is_termination_msg=lambda x: "邮件已完成" in (x.get("content", "") or "")
        )
        
        groupchat = autogen.GroupChat(
            agents=[user_proxy, self.agents["email"]],
            messages=[],
            max_round=5
        )
        manager = autogen.GroupChatManager(
            groupchat=groupchat,
            llm_config=self.llm_config
        )
        
        user_proxy.initiate_chat(manager, message=task)
        return self._extract_last_message(groupchat)
    
    def report_workflow(self, task: str) -> str:
        """数据分析工作流：搜索数据→分析→撰写报告→审查"""
        user_proxy = autogen.UserProxyAgent(
            name="用户",
            human_input_mode="NEVER",
            max_consecutive_auto_reply=15,
            is_termination_msg=lambda x: "审查通过" in (x.get("content", "") or "")
        )
        
        groupchat = autogen.GroupChat(
            agents=[
                user_proxy,
                self.agents["search"],
                self.agents["data"],
                self.agents["doc"],
                self.agents["reviewer"]
            ],
            messages=[],
            max_round=20,
            speaker_selection_method="round_robin"
        )
        manager = autogen.GroupChatManager(
            groupchat=groupchat,
            llm_config=self.llm_config
        )
        
        user_proxy.initiate_chat(manager, message=task)
        return self._extract_last_message(groupchat)
    
    def _extract_last_message(self, groupchat) -> str:
        """提取最后一条助手消息"""
        for msg in reversed(groupchat.messages):
            if msg["role"] == "assistant" and msg.get("content"):
                return msg["content"]
        return "无输出"

    def run(self, task_type: str, task: str) -> str:
        """执行办公任务"""
        workflows = {
            "email": self.email_workflow,
            "report": self.report_workflow,
            "document": self.report_workflow  # 复用报告流程
        }
        
        workflow = workflows.get(task_type)
        if not workflow:
            return f"不支持的任务类型: {task_type}"
        
        return workflow(task)

1.3.4 FastAPI网关

python

# api/gateway.py
"""API网关"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn

app = FastAPI(title="智能办公系统 API")

# 工作流实例（全局单例）
workflow = None

class TaskRequest(BaseModel):
    """任务请求"""
    task_type: str  # email / report / document
    description: str
    user_id: Optional[str] = None

class TaskResponse(BaseModel):
    """任务响应"""
    task_id: str
    status: str
    result: Optional[str] = None

@app.on_event("startup")
async def startup():
    """初始化工作流"""
    global workflow
    llm_config = {
        "config_list": [
            {
                "model": "Qwen/Qwen2.5-7B-Instruct",
                "base_url": "http://vllm-service:8000/v1",
                "api_key": "not-needed"
            }
        ]
    }
    workflow = OfficeWorkflow(llm_config)

@app.post("/api/tasks", response_model=TaskResponse)
async def create_task(request: TaskRequest):
    """创建办公任务"""
    import uuid
    task_id = str(uuid.uuid4())[:8]
    
    try:
        result = workflow.run(request.task_type, request.description)
        return TaskResponse(
            task_id=task_id,
            status="completed",
            result=result
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/health")
async def health():
    return {"status": "healthy", "service": "smart-office"}

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

1.3.5 Streamlit前端界面

python

# ui/app.py
"""智能办公系统前端"""

import streamlit as st
import requests
import json

st.set_page_config(page_title="智能办公系统", layout="wide")

st.title("🏢 多Agent协作智能办公系统")

# 侧边栏
with st.sidebar:
    st.header("任务类型")
    task_type = st.selectbox(
        "选择办公任务",
        ["email", "report", "document"],
        format_func=lambda x: {
            "email": "📧 邮件撰写",
            "report": "📊 数据分析报告",
            "document": "📝 文档撰写"
        }[x]
    )
    
    st.header("关于")
    st.info("""
    本系统使用多Agent协作自动完成办公任务。
    - 📧 邮件Agent：撰写和回复邮件
    - 📊 数据Agent：数据分析与报表
    - 📝 文档Agent：文档撰写与审查
    - 🔍 搜索Agent：信息检索
    - ✅ 审查Agent：质量把控
    """)

# 主区域
st.header(f"📋 任务描述")

task_description = st.text_area(
    "请输入您的任务需求",
    height=200,
    placeholder="例如：给张总写一封邮件，汇报本周项目进度..."
)

col1, col2 = st.columns([1, 5])
with col1:
    submit = st.button("🚀 提交任务", type="primary", use_container_width=True)

if submit and task_description:
    with st.spinner("🤖 多Agent正在协作处理中..."):
        try:
            response = requests.post(
                "http://localhost:8000/api/tasks",
                json={
                    "task_type": task_type,
                    "description": task_description
                },
                timeout=120
            )
            
            if response.status_code == 200:
                result = response.json()
                
                st.success("✅ 任务完成！")
                
                with st.expander("📄 查看结果", expanded=True):
                    st.markdown(result["result"])
            else:
                st.error(f"任务失败: {response.text}")
                
        except Exception as e:
            st.error(f"连接失败: {str(e)}")

elif submit:
    st.warning("请先输入任务描述")

# 历史记录
st.divider()
st.header("📜 历史记录")
st.info("此处可扩展显示历史任务记录")

1.4 项目目录结构

smart-office/
├── README.md                    # 项目说明文档
├── requirements.txt             # Python依赖
├── docker-compose.yml           # Docker编排
├── api/
│   ├── __init__.py
│   └── gateway.py              # FastAPI网关
├── agents/
│   ├── __init__.py
│   └── work_agents.py          # Agent定义
├── workflows/
│   ├── __init__.py
│   └── office_workflow.py      # 工作流定义
├── rag/
│   ├── __init__.py
│   └── knowledge_base.py       # RAG知识库
├── ui/
│   ├── __init__.py
│   └── app.py                  # Streamlit前端
├── data/
│   └── knowledge/              # 知识库文档
└── tests/
    ├── test_agents.py
    ├── test_workflow.py
    └── test_api.py

项目二：企业级AI客服中台

2.1 项目概述

项目名称：Enterprise AI Customer Service Platform

技术栈：FastAPI + RAG + Agent + ChromaDB + Redis + Docker

项目目标：构建一个企业级AI客服中台，具备以下核心功能：

🤖 智能问答：基于企业知识库的FAQ自助服务
🔄 多轮对话：维护对话上下文，处理复杂问题
👨‍💼 人工转接：智能识别需要人工处理的情况
📊 数据分析：客服质量仪表盘
⚙️ 后台管理：知识库管理、Agent配置

2.2 系统架构

                        用户入口
                           │
              ┌────────────┴────────────┐
              │    Nginx 负载均衡        │
              └────────────┬────────────┘
                           │
              ┌────────────┴────────────┐
              │   FastAPI 服务集群       │
              │  ├─ 意图识别 Agent       │
              │  ├─ FAQ回复 Agent        │
              │  ├─ 工单生成 Agent       │
              │  └─ 坐席助手 Agent       │
              └────────┬────────┬───────┘
                       │        │
              ┌────────┘        └────────┐
              │                          │
         ┌────┴─────┐            ┌──────┴──────┐
         │ ChromaDB  │            │   Redis     │
         │ 知识库向量  │            │  会话缓存   │
         └──────────┘            └─────────────┘
              │
         ┌────┴─────┐
         │  MySQL    │
         │ 业务数据   │
         └──────────┘

2.3 核心模块实现

2.3.1 客服核心引擎

python

# core/engine.py
"""AI客服核心引擎"""

from openai import OpenAI
import json
from enum import Enum
from typing import Optional

class CustomerIntent(str, Enum):
    """客户意图"""
    GREETING = "greeting"
    FAQ = "faq"
    COMPLAINT = "complaint"
    ORDER = "order"
    TECHNICAL = "technical"
    HUMAN = "human"

class CustomerServiceEngine:
    """AI客服引擎"""
    
    def __init__(self, knowledge_base=None, model="gpt-4"):
        self.client = OpenAI()
        self.kb = knowledge_base
        self.model = model
    
    def analyze_intent(self, message: str, history: list = None) -> dict:
        """意图分析"""
        prompt = f"""消息：{message}
历史：{json.dumps(history[-5:], ensure_ascii=False) if history else '无'}

分析意图，输出JSON：
{{
    "intent": "greeting|faq|complaint|order|technical|human",
    "confidence": 0.0-1.0,
    "sub_topics": ["关键词1", "关键词2"],
    "sentiment": "positive|neutral|negative|angry",
    "needs_human": true/false
}}
"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def generate_reply(self, message: str, intent: dict, context: str = "") -> str:
        """生成回复"""
        system_prompt = f"""你是一家智能客服机器人。当前客户意图：{intent['intent']}
情绪：{intent.get('sentiment', 'neutral')}

回复原则：
1. 友好、专业、简洁（不超过200字）
2. 如果是投诉，先道歉再解决问题
3. 如果无法解决，主动提供转人工选项
4. 每次回复后问一句是否需要进一步帮助

{context if context else ''}"""
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": message}
            ],
            max_tokens=500
        )
        return response.choices[0].message.content
    
    def retrieve_knowledge(self, query: str, k: int = 3) -> str:
        """检索知识库"""
        if not self.kb:
            return ""
        results = self.kb.search(query, k=k)
        return "\n".join([
            f"[{r['source']}] {r['content']}"
            for r in results
        ])
    
    def handle(self, message: str, session_id: str, history: list = None) -> dict:
        """完整处理流程"""
        # 1. 意图分析
        intent = self.analyze_intent(message, history)
        print(f"[意图] {intent['intent']} (置信度: {intent['confidence']:.2f})")
        
        # 2. 检索相关知识
        context = self.retrieve_knowledge(message)
        
        # 3. 生成回复
        reply = self.generate_reply(message, intent, context)
        
        # 4. 判断是否需要转人工
        needs_human = (
            intent.get("needs_human", False) or
            intent["intent"] == CustomerIntent.HUMAN or
            intent.get("sentiment") == "angry"
        )
        
        return {
            "reply": reply,
            "intent": intent["intent"],
            "confidence": intent["confidence"],
            "needs_human": needs_human,
            "context_used": bool(context)
        }

2.3.2 会话管理（Redis）

python

# core/session.py
"""会话管理"""

import redis
import json
import uuid
from datetime import datetime
from typing import Optional

class SessionManager:
    """Redis会话管理器"""
    
    def __init__(self, redis_url="redis://localhost:6379/0"):
        self.redis = redis.from_url(redis_url)
        self.session_ttl = 3600  # 1小时过期
    
    def create_session(self, user_id: str) -> str:
        """创建新会话"""
        session_id = str(uuid.uuid4())[:8]
        session = {
            "session_id": session_id,
            "user_id": user_id,
            "created_at": datetime.now().isoformat(),
            "messages": [],
            "context": {},
            "status": "active"
        }
        self.redis.setex(
            f"session:{session_id}",
            self.session_ttl,
            json.dumps(session, ensure_ascii=False)
        )
        return session_id
    
    def add_message(self, session_id: str, role: str, content: str):
        """添加对话消息"""
        session_data = self.redis.get(f"session:{session_id}")
        if not session_data:
            return None
        
        session = json.loads(session_data)
        session["messages"].append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
        session["updated_at"] = datetime.now().isoformat()
        
        # 保留最近20条消息
        if len(session["messages"]) > 20:
            session["messages"] = session["messages"][-20:]
        
        self.redis.setex(
            f"session:{session_id}",
            self.session_ttl,
            json.dumps(session, ensure_ascii=False)
        )
        return session
    
    def get_history(self, session_id: str, last_n: int = 5) -> list:
        """获取最近N条历史消息"""
        session_data = self.redis.get(f"session:{session_id}")
        if not session_data:
            return []
        
        session = json.loads(session_data)
        return session["messages"][-last_n:]
    
    def close_session(self, session_id: str):
        """关闭会话"""
        session_data = self.redis.get(f"session:{session_id}")
        if session_data:
            session = json.loads(session_data)
            session["status"] = "closed"
            session["closed_at"] = datetime.now().isoformat()
            self.redis.setex(
                f"session:{session_id}",
                self.session_ttl,
                json.dumps(session, ensure_ascii=False)
            )

2.3.3 坐席助手模块

python

# core/agent_assistant.py
"""坐席助手"""

from openai import OpenAI

class AgentAssistant:
    """坐席助手：为人工客服提供实时支持"""
    
    def __init__(self, model="gpt-4"):
        self.client = OpenAI()
        self.model = model
    
    def analyze_conversation(self, history: list) -> dict:
        """分析对话并生成坐席建议"""
        conversation = "\n".join([
            f"{msg['role']}: {msg['content']}"
            for msg in history[-10:]
        ])
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": """你是一名AI坐席助手，分析对话并输出以下JSON：

{
    "summary": "对话摘要（一句话）",
    "customer_sentiment": "积极/中性/不满/愤怒",
    "issue_type": "问题分类",
    "suggested_reply": "建议坐席回复的话术",
    "knowledge_tips": ["相关知识点1", "相关知识点2"],
    "escalation_needed": true/false,
    "urgency": "low/medium/high/urgent"
}"""},
                {"role": "user", "content": f"对话记录：\n{conversation}"}
            ],
            response_format={"type": "json_object"}
        )
        return response.choices[0].message.content
    
    def generate_knowledge_card(self, topic: str) -> str:
        """生成知识卡片"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是一名知识库编辑。请为坐席生成一个简洁的知识卡片，包含：问题、原因、解决方案、注意事项。"},
                {"role": "user", "content": f"主题：{topic}"}
            ]
        )
        return response.choices[0].message.content

2.3.4 完整API服务

python

# main.py
"""AI客服中台主服务"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import uvicorn

from core.engine import CustomerServiceEngine
from core.session import SessionManager
from core.agent_assistant import AgentAssistant
from rag.knowledge_base import EnterpriseKnowledgeBase

# ========= 数据模型 =========
class ChatRequest(BaseModel):
    """聊天请求"""
    message: str
    session_id: Optional[str] = None
    user_id: str = "anonymous"

class ChatResponse(BaseModel):
    """聊天响应"""
    reply: str
    session_id: str
    intent: str
    needs_human: bool

class SessionInfo(BaseModel):
    """会话信息"""
    session_id: str
    message_count: int
    status: str

# ========= FastAPI应用 =========
app = FastAPI(title="企业AI客服中台")

# 全局实例
engine = None
session_mgr = None
assistant = None

@app.on_event("startup")
async def startup():
    global engine, session_mgr, assistant
    
    # 初始化知识库
    kb = EnterpriseKnowledgeBase("./kb_store")
    kb.load()
    
    # 初始化引擎
    engine = CustomerServiceEngine(knowledge_base=kb)
    session_mgr = SessionManager()
    assistant = AgentAssistant()

@app.post("/api/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """对话接口"""
    # 获取或创建会话
    session_id = request.session_id
    if not session_id:
        session_id = session_mgr.create_session(request.user_id)
    
    # 获取历史
    history = session_mgr.get_history(session_id)
    
    # 处理消息
    result = engine.handle(request.message, session_id, history)
    
    # 保存消息
    session_mgr.add_message(session_id, "user", request.message)
    session_mgr.add_message(session_id, "assistant", result["reply"])
    
    return ChatResponse(
        reply=result["reply"],
        session_id=session_id,
        intent=result["intent"],
        needs_human=result["needs_human"]
    )

@app.post("/api/sessions", response_model=SessionInfo)
async def create_session(user_id: str = "anonymous"):
    """创建新会话"""
    session_id = session_mgr.create_session(user_id)
    return SessionInfo(
        session_id=session_id,
        message_count=0,
        status="active"
    )

@app.post("/api/agent-assist")
async def agent_assist(session_id: str):
    """坐席助手接口"""
    history = session_mgr.get_history(session_id, last_n=20)
    if not history:
        raise HTTPException(status_code=404, detail="会话不存在")
    
    analysis = assistant.analyze_conversation(history)
    return {"analysis": analysis}

@app.get("/api/health")
async def health():
    return {"status": "healthy", "service": "ai-customer-service"}

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        workers=4,
        reload=False
    )

2.4 部署方案

yaml

# docker-compose.yml
version: '3.8'

services:
  # API服务
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - redis
      - chromadb
    restart: unless-stopped
    deploy:
      replicas: 3
      resources:
        limits:
          memory: "2G"
          cpus: "2"
  
  # Redis缓存
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped
  
  # ChromaDB向量数据库
  chromadb:
    image: chromadb/chroma:latest
    ports:
      - "8001:8000"
    volumes:
      - chroma_data:/chroma/chroma
    environment:
      - IS_PERSISTENT=TRUE
    restart: unless-stopped
  
  # Nginx负载均衡
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
    depends_on:
      - api
    restart: unless-stopped

volumes:
  redis_data:
  chroma_data:

2.5 项目目录结构

ai-customer-service/
├── README.md
├── requirements.txt
├── docker-compose.yml
├── nginx.conf
├── main.py                    # 主服务入口
├── core/
│   ├── __init__.py
│   ├── engine.py              # 客服引擎
│   ├── session.py             # 会话管理
│   └── agent_assistant.py     # 坐席助手
├── rag/
│   ├── __init__.py
│   └── knowledge_base.py     # 知识库
├── models/
│   └── schemas.py            # 数据模型
├── tools/
│   ├── kb_admin.py           # 知识库管理工具
│   └── analytics.py          # 数据分析工具
├── tests/
│   ├── test_engine.py
│   ├── test_session.py
│   └── test_api.py
└── data/
    ├── knowledge/            # 知识库原始文档
    └── logs/                 # 对话日志

项目交付标准

功能要求

类别	要求	分值
核心功能	所有Agent正常工作，任务可完成	40%
架构设计	模块化、可扩展、符合设计模式	20%
代码质量	类型注解、文档字符串、错误处理	15%
测试覆盖	单元测试覆盖核心逻辑	10%
部署文档	Docker/docker-compose一键部署	10%
演示视频	3-5分钟功能演示	5%

加分项

集成真实LLM API（OpenAI/通义千问）
前端界面美化（使用Streamlit/React）
添加日志和监控
性能测试报告（包含QPS和延迟数据）
用户认证和权限控制

小结与练习

知识小结

项目一核心：AutoGen多Agent协作 + LangChain RAG + vLLM推理
项目二核心：FastAPI + Redis会话 + 意图识别 + 坐席助手
通用能力：架构设计、模块拆分、接口规范、Docker部署

练习

练习1：多Agent扩展

在项目一基础上增加一个新Agent：
1. 日历Agent：管理日程和会议安排
2. 任务Agent：跟踪项目进度和待办事项
实现Agent定义、工作流集成和前端展示。

练习2：客服数据分析

为项目二添加数据分析功能：
1. 每日对话量统计
2. 意图分布分析
3. 转人工率统计
4. 客户满意度分析
使用图表展示（Matplotlib/ECharts）。

练习3：性能压测

使用locust或wrk对项目进行压力测试：
1. 测试QPS和延迟性能
2. 找出性能瓶颈
3. 提出优化方案
4. 生成性能测试报告

练习4：安全加固

为系统添加安全功能：
1. API认证（JWT Token）
2. 请求频率限制
3. 敏感词过滤
4. 对话内容脱敏
5. 审计日志

恭喜你完成了模块五的全部学习！🎉 你已掌握了从Agent核心原理到企业级应用部署的完整知识体系。

章节5：结业项目实战 ​

本章导学 ​

学习目标 ​

项目一：多Agent协作智能办公系统 ​

1.1 项目概述 ​

1.2 系统架构 ​

1.3 技术方案详解 ​

1.3.1 Agent定义 ​

1.3.2 RAG知识库（LangChain + ChromaDB） ​

1.3.3 AutoGen工作流 ​

1.3.4 FastAPI网关 ​

1.3.5 Streamlit前端界面 ​

1.4 项目目录结构 ​

项目二：企业级AI客服中台 ​

2.1 项目概述 ​

2.2 系统架构 ​

2.3 核心模块实现 ​

2.3.1 客服核心引擎 ​

2.3.2 会话管理（Redis） ​

2.3.3 坐席助手模块 ​

2.3.4 完整API服务 ​

2.4 部署方案 ​

2.5 项目目录结构 ​

项目交付标准 ​

功能要求 ​

加分项 ​

小结与练习 ​

知识小结 ​

练习 ​

练习1：多Agent扩展 ​

练习2：客服数据分析 ​

练习3：性能压测 ​

练习4：安全加固 ​

章节5：结业项目实战

本章导学

学习目标

项目一：多Agent协作智能办公系统

1.1 项目概述

1.2 系统架构

1.3 技术方案详解

1.3.1 Agent定义

1.3.2 RAG知识库（LangChain + ChromaDB）

1.3.3 AutoGen工作流

1.3.4 FastAPI网关

1.3.5 Streamlit前端界面

1.4 项目目录结构

项目二：企业级AI客服中台

2.1 项目概述

2.2 系统架构

2.3 核心模块实现

2.3.1 客服核心引擎

2.3.2 会话管理（Redis）

2.3.3 坐席助手模块

2.3.4 完整API服务

2.4 部署方案

2.5 项目目录结构

项目交付标准

功能要求

加分项

小结与练习

知识小结

练习

练习1：多Agent扩展

练习2：客服数据分析

练习3：性能压测

练习4：安全加固