Python教程 Episode 24 - AI Agent高阶实践

欢迎来到「Python教程从零基础到实战」系列的第24期！

在前一期的我们深入学习了AI Agent的基础架构：一个能感知环境、做出决策并自主采取行动的智能程序。我们已经搭建了自己的天气查询Agent、学习了LangChain框架，也理解了ReAct循环——思考→行动→观察。

但是，现实世界中的问题往往更加复杂。单个Agent可能能力有限，无法独立完成整个任务链条；又或者我们需要多个Agent之间互相协作，才能完成像"市场调研报告生成"这样综合性的工作。

今天，我们要把视野拉大一点，探讨 **AI Agent的高阶实践**。让我们一起解锁多Agent协作、AutoGen框架、RAG增强智能体，以及如何将Agent部署为Web服务的技能树。

---

## 第一部分：为什么需要多Agent系统？

想象一下，你要写一份关于"2025年全球电动汽车市场"的综合报告。这个过程包含：

1.**搜索**收集最新数据和新闻

2.**分析**整理数据图表

3.**写作**撰写分析报告

4.**审核**检查内容准确性

5.**格式**排版最终文档

如果让单个Agent完成所有步骤，它可能会在每一步上都做得平平无奇。但如果让每个Agent专注自己的领域，效果会怎样呢？

```

🔍 研究员Agent → 搜索数据

📊 分析师Agent → 处理数据

✍️ 作家Agent → 撰写文章

👨‍⚖️ 审核员Agent → 质量把关

🎨 设计师Agent → 排版美化

```

这就是**多Agent系统**的核心价值：**分而治之，各展所长**。通过合理的分工和协作，每个Agent都能在自己的专业领域发挥最大价值，整体产出远高于单个Agent的简单叠加。

---

## 第二部分：使用 AutoGen 构建多 Agent 协作

微软开发的 [AutoGen](https://microsoft.github.io/autogen/) 是一个强大的框架，专为多Agent协作而设计。它的核心思想是：**让不同的Agent（角色）通过自然语言相互对话，共同完成任务**。

让我们先看一个简单的例子：创建一个代码审查系统。

### 2.1 安装依赖

```bash

pipinstallautoagen[openai]

```

> ⚠️ 注意：正确的包名是 `auto-gen`，实际安装命令为：

> ```bash

>pip install auto-gen openai python-dotenv

> ```

### 2.2 定义两个Agent：开发者和审查者

```python

import os

from dotenv import load_dotenv

from autogen import AssistantAgent, UserProxyAgent, configure_conv_manual_turns

from openai import OpenAI

load_dotenv()

# 初始化客户端

client = OpenAI(

api_key=os.getenv("OPENAI_API_KEY"),

base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

)

# 1. 创建开发者Agent——负责编写代码

developer = AssistantAgent(

name="Developer",

llm_config={

"config_list": [

            {

"model": "gpt-4o-mini",

"api_key": os.getenv("OPENAI_API_KEY"),

"base_url": os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),

"seed": 42,

            }

        ],

    },

system_message="你是一个经验丰富的Python开发者，职责是编写干净、可维护的代码。遇到需求时先提问理解细节，然后提供代码实现。你的回复应简洁明了。"

)

# 2. 创建审查者Agent——负责代码审查

reviewer = AssistantAgent(

name="Reviewer",

llm_config={

"config_list": [

            {

"model": "gpt-4o-mini",

"api_key": os.getenv("OPENAI_API_KEY"),

"base_url": os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),

"seed": 42,

            }

        ],

    },

system_message="你是一个资深的代码审查专家，专注于发现代码中的潜在问题：性能瓶颈、安全漏洞、可读性差的地方。不要直接修改代码，而是指出问题并提供改进建议。坚持高标准。"

)

# 3. 创建用户代理——作为人工介入点

user_proxy = UserProxyAgent(

name="User",

code_execution_config=False,  # 禁用代码执行确保安全

is_human=True,

)

# 启动对话回路

print("=== 多Agent协作：代码审查流程 ===\n")

configure_conv_manual_turns(user_proxy, [developer, reviewer])

# 用户提出需求

user_proxy.initiate_chat(

    developer,

message="请写一个函数，接收一个整数列表，返回其中所有偶数的平方和。要求代码简洁高效，有类型提示和注释。",

)

# 等待审查者的反馈

input("\n按回车键继续查看审查结果...")

# 审查者直接向开发者提问

reviewer.send(

"我注意到你的实现没有处理空列表的情况，请补充说明或增加边界测试。",

recipient=developer,

)

# 开发者回应

developer.send(

"很好的意见！我已经添加了边界检查和详细的docstring。现在代码更健壮了。请看：\n\ndef even_square_sum(numbers: list[int]) -> int:\n\"\"\"计算列表中所有偶数的平方和。\n

    Args:

        numbers: 整数列表

    Returns:

        所有偶数的平方和。如果列表为空或没有偶数，返回0。

    \"\"

ifnot numbers:

return0

returnsum(x**2for x in numbers if x % 2 == 0)",

recipient=user_proxy,

)

print("\n✅ 协作完成：开发者与审查者通过对话产出了高质量的代码。")

```

在这个例子中：

-`Developer` 专注编码实现

-`Reviewer` 专注质量保证

-`User` 作为仲裁者和最终验收者

三个Agent通过**自然语言对话**完成了整个流程，而不需要任何硬编码的逻辑控制。这就是多Agent系统的魅力所在。

---

## 第三部分：RAG增强Agent——让Agent拥有知识库

前面的Agent虽然聪明，但它们的知识仅限于训练数据截止时的内容。对于企业应用来说，我们通常希望Agent能够访问**公司内部的特定知识**——产品手册、客户合同、技术文档等。

这就引入了 **RAG（Retrieval-Augmented Generation，检索增强生成）** 的概念：**在使用LLM之前，先从相关文档中检索出最匹配的信息，把这些信息作为上下文提供给LLM**。

### 3.1 RAG的工作流程

```

用户提问

   ↓

【检索器】从向量数据库中查找最相关的文档片段

   ↓

【组装器】将检索到的片段 + 原始问题组合成新的Prompt

   ↓

【LLM】基于增强的上下文生成回答

   ↓

输出回答

```

### 3.2 实战：构建带有产品知识库的客服Agent

我们将使用 `chromadb` 作为本地向量数据库，`langchain` 作为RAG框架。

#### 安装依赖

```bash

pipinstalllangchainchromadbsentence-transformerspython-dotenv

```

#### 准备知识库文件

首先创建一个产品文档目录 `knowledge_base/`，并放入以下文件：

**products.txt**:

```

我们的主要产品包括：

1. SmartHome Hub（智能家庭中枢）

   - 功能：控制家中的所有智能设备

   - 特点：支持Wi-Fi、Zigbee、蓝牙三种协议

   - 价格：¥299

   - 保修期：2年

2. SecurityCam Pro（专业安防摄像头）

   - 功能：高清夜视监控，移动检测警报

   - 特点：1080P分辨率，夜视距离10米

   - 价格：¥499

   - 保修期：1年

3. EnergyMonitor Plus（能耗监测仪）

   - 功能：实时监测家庭用电情况

   - 特点：USB接口，APP数据可视化

   - 价格：¥199

   - 保修期：1年

```

**faq.txt**:

```

常见问题：

Q: 产品的保修政策是什么？

A: SmartHome Hub享受2年保修，其他产品1年保修。保修期内非人为故障免费维修。

Q: 如何连接Zigbee设备？

A: 打开SmartHome Hub的配对模式，确保Zigbee设备处于可被发现状态，两者会自动连接。

Q: APP支持哪些操作系统？

A: iOS 12以上和Android 8.0以上都支持。可以在App Store和各大安卓应用市场下载。

Q: 网络连接失败怎么办？

A: 请确认路由器正常，设备在Wi-Fi范围内，重启尝试重新连接。如果问题持续，联系客服。

```

#### 实现RAG Agent

```python

from langchain_community.vectorstores import Chroma

from langchain_text_splitters import CharacterTextSplitter

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

from langchain.chains import RetrievalQA

from langchain_core.prompts import PromptTemplate

from dotenv import load_dotenv

import os

import time

load_dotenv()

# 1. 加载并分割文档

defload_and_split_documents():

"""从文本文件中加载文档并进行分割"""

    documents = []

# 读取产品文档

withopen('knowledge_base/products.txt', 'r', encoding='utf-8') as f:

        documents.append(("products.txt", f.read()))

# 读取FAQ文档

withopen('knowledge_base/faq.txt', 'r', encoding='utf-8') as f:

        documents.append(("faq.txt", f.read()))

# 使用字符分割器

    text_splitter = CharacterTextSplitter(

chunk_size=1000,      # 每个chunk的最大长度

chunk_overlap=200,    # chunk之间的重叠长度

separator="\n\n"# 分割符

    )

    all_chunks = []

for filename, content in documents:

        chunks = text_splitter.split_text(content)

for i, chunk inenumerate(chunks):

            all_chunks.append({

"page_content": chunk,

"metadata": {"source": filename, "chunk_num": i + 1}

            })

return all_chunks

# 2. 创建向量索引

defcreate_vector_index(documents):

"""将文本块转换为向量并存储在ChromaDB中"""

print("正在创建向量索引...")

    embeddings = OpenAIEmbeddings(

api_key=os.getenv("OPENAI_API_KEY"),

base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

    )

# 存储文档的列表格式

    texts = [doc["page_content"] for doc in documents]

    metadatas = [doc["metadata"] for doc in documents]

    vector_db = Chroma.from_texts(

texts=texts,

metadatas=metadatas,

embedding=embeddings,

persist_directory="./rag_vector_store"# 持久化到磁盘

    )

print(f"✅ 向量索引创建成功！共存储 {len(vectors)} 个向量。")

return vector_db

# 3. 创建带RAG的Agent

classRagAgent:

"""结合了检索能力的Agent"""

def__init__(self, vector_db, model_name="gpt-4o-mini"):

self.vector_db = vector_db

self.llm = ChatOpenAI(

model=model_name,

api_key=os.getenv("OPENAI_API_KEY"),

base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),

temperature=0.3

        )

# 定制Prompt模板

self.prompt_template = PromptTemplate.from_template(

"""你是一名专业的客户支持助手，拥有公司全部产品文档和FAQ知识库。

            ### 检索到的知识库内容：

{context}

            ### 用户问题：

{question}

            ### 要求：

            1. 只在知识库范围内回答问题，不要编造不存在的信息。

            2. 如果知识库中没有相关信息，诚实地告诉用户你无法回答，并建议查阅官方文档或联系客户服务。

            3. 回答时使用清晰、友好的语气，包含必要细节。

            4. 如果问题涉及多个产品，分别说明。

            现在请回答用户的问题：

            """

        )

defask(self, question):

"""向Agent提问并使用RAG获取答案"""

# 从向量库中检索相关内容

        relevant_docs = self.vector_db.similarity_search(question, k=3)

# 构造上下文

        context = "\n\n---\n\n".join([

f"来源：{doc['metadata']['source']} (第{doc['metadata']['chunk_num']}段)\n{doc['page_content']}"

for doc in relevant_docs

        ])

# 使用定制的Prompt

        prompt = self.prompt_template.format(context=context, question=question)

# 调用LLM

        response = self.llm.invoke(prompt)

return response.content, relevant_docs

# === 主程序 ===

if__name__ == "__main__":

# 如果向量库已经存在，直接加载；否则重新创建

try:

print("🔄 尝试加载已有的向量索引...")

        vector_db = Chroma(persist_directory="./rag_vector_store", 

embedding_function=OpenAIEmbeddings(

api_key=os.getenv("OPENAI_API_KEY"),

base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")))

print("✅ 成功加载现有向量索引。")

except:

print("❌ 未找到现有索引，正在创建新的...")

        docs = load_and_split_documents()

        vector_db = create_vector_index(docs)

# 创建Agent实例

    agent = RagAgent(vector_db)

# 交互式问答界面

print("\n" + "="*60)

print(" 🤖 客户支持RAG Agent — 输入 'quit' 退出")

print("="*60)

whileTrue:

        user_input = input("\n👤 你: ").strip()

if user_input.lower() in ("quit", "exit", "退出"):

print("🤖 Agent: 感谢使用，再见！")

break

ifnot user_input:

continue

print("\n🤖 Agent: 请稍候正在搜索知识库...")

        start_time = time.time()

        answer, docs = agent.ask(user_input)

        elapsed = time.time() - start_time

print(f"\n🤖 Agent: {answer}")

print(f"\n（检索耗时: {elapsed:.2f}秒）")

# 显示检索到的来源

if docs:

print("\n📚 参考来源:")

for doc in docs:

                source = doc.metadata["source"]

                chunk = doc.metadata["chunk_num"]

print(f"   • {source} 第{chunk}段")

```

运行这个脚本后，你可以尝试问一些关于产品的问题，比如：

```

👤 你: SmartHub的价格是多少？

👤 你: 如何连接Zigbee设备？

👤 你: 你们的保修政策是怎样的？

👤 你: 这个产品支持语音控制吗？（测试不在知识库内的问题）

```

Agent会从知识库中检索相关信息并给出准确回答。如果知识库中没有的答案，它会诚实地告诉你不知道——这正是RAG的价值所在：**避免幻觉，基于事实回答**。

---

## 第四部分：将Agent部署为Web服务

当你的Agent开发完成后，如何将它提供给用户使用呢？最常见的方式就是部署为一个**Web API**，前端或其他应用程序可以通过HTTP请求来调用它。

本部分我们将使用 **FastAPI**（一个现代的快速Python Web框架）来构建一个RESTful API服务，让用户可以通过浏览器或手机应用与Agent交互。

### 4.1 项目结构

```

agent_web_service/

├── main.py              # FastAPI应用入口

├── agent.py             # Agent核心逻辑

├── requirements.txt     # 依赖包

└── .env               # 环境变量配置

```

### 4.2 Agent核心逻辑（agent.py）

我们将把之前的RAG Agent封装成一个可复用的类：

```python

# agent.py

from typing import List, Dict, Any

from langchain_community.vectorstores import Chroma

from langchain_text_splitters import CharacterTextSplitter

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

from dotenv import load_dotenv

import os

import json

load_dotenv()

classKnowledgeBasedAgent:

"""基于知识库的Agent，可被Web服务调用"""

def__init__(self, vector_store_path: str = "./rag_vector_store"):

self.vector_store_path = vector_store_path

self._load_vector_db()

self.llm = ChatOpenAI(

model=os.getenv("AGENT_MODEL_NAME", "gpt-4o-mini"),

api_key=os.getenv("OPENAI_API_KEY"),

base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),

temperature=0.3

        )

self.system_prompt = """你是一名专业的客户支持助手，拥有公司全部产品文档和FAQ知识库。

请基于提供的知识库内容回答问题，只使用已知信息，不编造内容。"""

def_load_vector_db(self):

"""加载或初始化向量数据库"""

try:

from langchain_community.vectorstores import Chroma

from langchain_openai import OpenAIEmbeddings

self.db = Chroma(

persist_directory=self.vector_store_path,

embedding_function=OpenAIEmbeddings(

api_key=os.getenv("OPENAI_API_KEY"),

base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

                )

            )

print(f"✅ 已加载向量数据库: {self.vector_store_path}")

exceptExceptionas e:

print(f"⚠️ 无法加载向量库，将使用内联知识: {e}")

self.db = None

defsearch_knowledge(self, query: str, k: int = 3) -> List[Dict[str, Any]]:

"""从知识库中检索相关内容"""

ifself.db isNone:

return []

        relevant_docs = self.db.similarity_search(query, k=k)

return [{

"page_content": doc.page_content,

"source": doc.metadata.get("source", "unknown"),

"chunk_num": doc.metadata.get("chunk_num", 1)

        } for doc in relevant_docs]

defgenerate_answer(self, query: str, knowledge_context: str) -> str:

"""基于检索到的知识生成答案"""

        prompt = f"""{self.system_prompt}

### 检索到的知识库内容：

{knowledge_context}

### 用户问题：

{query}

### 要求：

1. 只在知识库范围内回答问题，不要编造不存在的信息。

2. 如果知识库中没有相关信息，诚实地告诉用户。

3. 回答清晰、友好，包含必要细节。

现在请回答用户的问题：

"""

        response = self.llm.invoke(prompt)

return response.content

defprocess_query(self, query: str, top_k: int = 3) -> Dict[str, Any]:

"""处理用户查询并返回结构化结果"""

# 1. 检索知识库

        docs = self.search_knowledge(query, k=top_k)

# 2. 格式化上下文

        context_text = "\n\n".join([

f"【来源：{d['source']} 第{d['chunk_num']}段】\n{d['page_content'][:500]}..."

for d in docs

        ])

ifnot context_text:

            context_text = "知识库中暂无相关记录。"

# 3. 生成答案

        answer = self.generate_answer(query, context_text)

return {

"query": query,

"answer": answer,

"context_used": context_text,

"sources": [{"source": d["source"], "chunk": d["chunk_num"]} for d in docs],

"knowledge_retrieved": len(docs) > 0

        }

```

### 4.3 FastAPI服务（main.py）

```python

# main.py

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

from agent import KnowledgeBasedAgent

from dotenv import load_dotenv

import os

import time

load_dotenv()

# 初始化Agent（只在启动时加载一次）

agent = KnowledgeBasedAgent()

app = FastAPI(

title="Knowledge-Based Agent API",

version="1.0.0",

description="基于RAG的客户支持Agent REST API"

)

# Pydantic模型定义请求和响应

classQueryRequest(BaseModel):

    query: str = ""

    top_k: int = 3

classQueryResponse(BaseModel):

    query: str

    answer: str

    sources: list

    knowledge_retrieved: bool

    processing_time: float

@app.get("/health")

defhealth_check():

"""健康检查端点"""

return {

"status": "healthy",

"service": "knowledge-agent-api",

"timestamp": time.time()

    }

@app.post("/ask", response_model=QueryResponse)

asyncdefask_endpoint(request: QueryRequest):

"""主要查询端点——接受用户查询并返回答案"""

ifnot request.query orlen(request.query.strip()) == 0:

raise HTTPException(status_code=400, detail="查询内容不能为空")

    start_time = time.time()

try:

# 调用Agent处理查询

        result = agent.process_query(request.query, request.top_k)

        processing_time = time.time() - start_time

        result["processing_time"] = round(processing_time, 3)

return result

exceptExceptionas e:

raise HTTPException(

status_code=500, 

detail=f"处理查询时出错: {str(e)}"

        )

@app.get("/docs")

defcustom_docs():

"""自定义文档页面重定向到Swagger UI"""

return"""<html><head><meta http-equiv="refresh" content="0; url=/swagger-ui.html"></head></html>"""

if__name__ == "__main__":

import uvicorn

    host = os.getenv("API_HOST", "0.0.0.0")

    port = int(os.getenv("API_PORT", "8000"))

print(f"🚀 启动Agent Web服务: http://{host}:{port}")

print(f"📝 API文档: http://{host}:{port}/docs")

print(f"💪 健康检查: http://{host}/{port}/health")

    uvicorn.run(app, host=host, port=port)

```

### 4.4 运行与测试

#### 安装依赖（requirements.txt）

```txt

fastapi==0.112.0

uvicorn==0.30.1

python-dotenv==1.0.1

langchain==0.2.9

chromadb==0.5.0

sentence-transformers==2.3.1

openai==1.51.0

pydantic==2.8.0

```

#### 启动服务

```bash

# 设置环境变量

exportOPENAI_API_KEY="your-api-key"

exportOPENAI_BASE_URL="https://api.your-provider.com/v1"# 可选

exportAGENT_MODEL_NAME="gpt-4o-mini"

# 启动服务

pythonmain.py

```

访问 `http://localhost:8000/docs`，你会看到自动生成的API文档页面，可以直接在这里测试API。

#### 使用curl测试

```bash

# 正常的查询请求

curl-XPOST"http://localhost:8000/ask"\

  -H "Content-Type: application/json" \

-d'{

    "query": "SmartHub的价格是多少？",

    "top_k": 3

  }'

# 返回示例：

{

"query":"SmartHub的价格是多少？",

"answer":"SmartHome Hub（智能家庭中枢）的价格是¥299。",

"sources": [

    {"source":"products.txt","chunk":1}

  ],

"knowledge_retrieved":true,

"processing_time":2.456

}

```

#### 前端集成示例（HTML）

如果你想在网页中使用这个Agent，可以创建一个简单的前端页面：

```html

<!DOCTYPEhtml>

<html>

<head>

<title>RAG Agent Demo</title>

<style>

body { font-family: Arial, sans-serif; max-width: 800px; margin: 0auto; padding: 20px; }