实战案例开发
综合运用所学知识，开发三个完整的智能体应用实战案例
8.1 本章概览
在前面的章节中，我们系统学习了智能体开发的各个方面：从基础概念到工具调用，从记忆管理到多智能体协作。现在，让我们通过三个完整的实战案例，将所学知识融会贯通。
┌─────────────────────────────────────────────────────────────────┐ │ 本章案例总览 │ ├─────────────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ │ │ 案例1 │ │ 案例2 │ │ 案例3 │ │ │ │ 个人研究助手 │ │ 代码开发助手 │ │ 智能客服系统 │ │ │ │ │ │ │ │ │ │ │ │ • 自动搜索 │ │ • 需求理解 │ │ • 问题理解 │ │ │ │ • 笔记整理 │ │ • 方案设计 │ │ • 知识库检索 │ │ │ │ • 报告生成 │ │ • 代码编写 │ │ • 智能回复 │ │ │ │ │ │ • 测试修复 │ │ • 工单处理 │ │ │ │ [ReAct+记忆] │ │ [规划+工具] │ │ [多智能体] │ │ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ └─────────────────────────────────────────────────────────────────┘
每个案例都包含：需求分析、架构设计、组件选择、完整代码实现和流程图。建议跟随案例动手实践，加深理解。
8.2 案例1：个人研究助手
8.2.1 需求分析
作为学生或研究人员，经常需要深入研究某个课题。传统流程是：手动搜索资料 → 阅读整理 → 做笔记 → 写报告。这个过程耗时且容易遗漏重要信息。
目标：开发一个研究助手智能体，能够自动搜索相关资料、整理笔记、生成研究报告，大幅提高研究效率。
8.2.2 核心功能
- 智能搜索 - 根据研究主题自动搜索相关信息
- 内容提取 - 从网页/文档中提取关键内容
- 笔记管理 - 自动分类整理笔记，支持检索
- 报告生成 - 基于收集的资料生成结构化报告
- 问答支持 - 基于已收集资料回答相关问题
8.2.3 系统架构
个人研究助手架构
┌──────────────────────────────────────────────────────────────┐
│ │
│ 用户输入 │
│ │ "研究气候变化对农业的影响" │
│ ▼ │
│ ┌──────────────────────────────────────────┐ │
│ │ ReAct 循环 │ │
│ │ ┌─────────┐ ┌─────────┐ ┌────────┐│ │
│ │ │ 思考 │───►│ 行动 │───►│ 观察 ││ │
│ │ │ Thought │ │ Action │ │Observe ││ │
│ │ └─────────┘ └────┬────┘ └────┬───┘│ │
│ │ │ │ │ │
│ │ ▼ │ │ │
│ │ ┌──────────────────────────────────┐ │ │
│ │ │ 工具层 │ │ │
│ │ │ ┌────────┐ ┌────────┐ ┌────────┐│ │ │
│ │ │ │ 搜索 │ │ 网页 │ │ 笔记 ││ │ │
│ │ │ │ 工具 │ │ 抓取 │ │ 工具 ││ │ │
│ │ │ └────────┘ └────────┘ └────────┘│ │ │
│ │ └──────────────────────────────────┘ │ │
│ └──────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────────┐ │
│ │ 记忆系统 │ │
│ │ ┌────────────┐ ┌────────────┐ │ │
│ │ │ 短期记忆 │ │ 长期记忆 │ │ │
│ │ │ (上下文) │ │ (笔记库) │ │ │
│ │ └────────────┘ └────────────┘ │ │
│ └──────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 生成研究报告 │
│ │
└──────────────────────────────────────────────────────────────┘
8.2.4 完整代码实现
"""
个人研究助手 - Research Assistant Agent
功能：自动搜索、整理笔记、生成报告
架构：ReAct + 记忆管理
"""
import json
import asyncio
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import hashlib
@dataclass
class Note:
"""笔记数据结构"""
id: str
content: str
source: str # 来源URL或文档
category: str # 分类
tags: List[str] # 标签
created_at: str
relevance_score: float = 0.0 # 相关度评分
@dataclass
class ResearchTask:
"""研究任务"""
topic: str
sub_questions: List[str] # 子问题列表
collected_notes: List[Note]
status: str # pending/searching/analyzing/completed
report: Optional[str] = None
class SearchTool:
"""搜索工具 - 模拟网络搜索"""
def __init__(self, api_key: str = None):
self.api_key = api_key
async def search(self, query: str, num_results: int = 5) -> List[Dict]:
"""执行搜索，返回结果列表"""
# 实际实现中调用搜索引擎API（如SerpAPI、Google API等）
# 这里用模拟数据演示
mock_results = [
{
"title": f"关于'{query}'的研究结果 {i+1}",
"url": f"https://example.com/result{i+1}",
"snippet": f"这是关于{query}的第{i+1}条搜索结果摘要..."
}
for i in range(num_results)
]
print(f"[搜索] 查询: {query}, 找到 {len(mock_results)} 条结果")
return mock_results
async def fetch_content(self, url: str) -> str:
"""获取网页完整内容"""
# 实际实现中使用requests或爬虫工具
print(f"[抓取] 获取页面内容: {url}")
return f"页面 {url} 的完整内容..."
class NoteStorage:
"""笔记存储 - 长期记忆管理"""
def __init__(self, storage_path: str = "notes.json"):
self.storage_path = storage_path
self.notes: Dict[str, Note] = {}
self._load()
def _load(self):
"""从文件加载笔记"""
try:
with open(self.storage_path, 'r', encoding='utf-8') as f:
data = json.load(f)
for note_data in data:
note = Note(**note_data)
self.notes[note.id] = note
print(f"[存储] 已加载 {len(self.notes)} 条笔记")
except FileNotFoundError:
print("[存储] 创建新的笔记库")
def _save(self):
"""保存笔记到文件"""
data = [asdict(note) for note in self.notes.values()]
with open(self.storage_path, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def add(self, content: str, source: str, category: str,
tags: List[str], relevance: float = 0.0) -> Note:
"""添加新笔记"""
note_id = hashlib.md5(
f"{content}{source}{datetime.now()}".encode()
).hexdigest()[:8]
note = Note(
id=note_id,
content=content,
source=source,
category=category,
tags=tags,
created_at=datetime.now().isoformat(),
relevance_score=relevance
)
self.notes[note_id] = note
self._save()
print(f"[存储] 已保存笔记: {note_id}")
return note
def search(self, query: str, tags: List[str] = None) -> List[Note]:
"""搜索笔记"""
results = []
query_lower = query.lower()
for note in self.notes.values():
# 匹配内容或标签
content_match = query_lower in note.content.lower()
tag_match = tags is None or any(t in note.tags for t in tags)
if content_match and tag_match:
results.append(note)
# 按相关度排序
results.sort(key=lambda x: x.relevance_score, reverse=True)
return results
def get_by_category(self, category: str) -> List[Note]:
"""按分类获取笔记"""
return [n for n in self.notes.values() if n.category == category]
class ResearchAgent:
"""研究助手智能体 - 核心类"""
def __init__(self, llm_client, search_tool: SearchTool,
note_storage: NoteStorage):
self.llm = llm_client
self.search = search_tool
self.notes = note_storage
self.max_search_rounds = 3 # 最大搜索轮数
async def research(self, topic: str) -> ResearchTask:
"""
执行完整的研究流程
1. 分解研究问题
2. 搜索资料
3. 整理笔记
4. 生成报告
"""
print(f"\n{'='*60}")
print(f"开始研究主题: {topic}")
print(f"{'='*60}\n")
# 创建研究任务
task = ResearchTask(
topic=topic,
sub_questions=[],
collected_notes=[],
status="pending"
)
# 步骤1: 分解研究问题
task.sub_questions = await self._decompose_topic(topic)
print(f"[分析] 分解为 {len(task.sub_questions)} 个子问题:")
for i, q in enumerate(task.sub_questions, 1):
print(f" {i}. {q}")
# 步骤2: 搜索和收集资料
task.status = "searching"
for question in task.sub_questions:
await self._research_sub_question(task, question)
# 步骤3: 生成报告
task.status = "analyzing"
task.report = await self._generate_report(task)
task.status = "completed"
print(f"\n{'='*60}")
print("研究完成！")
print(f"{'='*60}")
return task
async def _decompose_topic(self, topic: str) -> List[str]:
"""将研究主题分解为子问题"""
prompt = f"""
请将以下研究主题分解为3-5个具体的子研究问题，
这些问题应该覆盖主题的主要方面。
研究主题: {topic}
请以JSON数组格式返回子问题列表，例如:
["子问题1", "子问题2", "子问题3"]
只返回JSON，不要有其他文字。
"""
response = await self.llm.acall(prompt)
try:
# 解析JSON响应
sub_questions = json.loads(response)
return sub_questions[:5] # 最多5个
except:
# 如果解析失败，使用默认分解
return [
f"{topic}的定义和背景是什么？",
f"{topic}的主要影响因素有哪些？",
f"{topic}的现状和发展趋势如何？",
f"{topic}面临的挑战和解决方案？"
]
async def _research_sub_question(self, task: ResearchTask,
question: str):
"""研究单个子问题"""
print(f"\n[研究子问题] {question}")
# 搜索相关资料
for round in range(self.max_search_rounds):
search_results = await self.search.search(question, num_results=3)
for result in search_results:
# 使用LLM判断信息价值
relevance = await self._evaluate_relevance(
result['snippet'], question
)
if relevance > 0.6: # 相关度阈值
# 获取详细内容
content = await self.search.fetch_content(result['url'])
# 保存笔记
note = self.notes.add(
content=content[:1000], # 截取前1000字符
source=result['url'],
category=task.topic,
tags=["research", task.topic, f"Q:{question[:20]}"],
relevance=relevance
)
task.collected_notes.append(note)
# 如果收集到足够资料，提前结束
if len(task.collected_notes) >= 5:
break
async def _evaluate_relevance(self, content: str,
question: str) -> float:
"""评估内容与问题的相关度 (0-1)"""
prompt = f"""
评估以下内容对回答问题的相关度。
问题: {question}
内容: {content}
请只返回一个0-1之间的数字，表示相关度。
1表示高度相关，0表示完全无关。
"""
try:
response = await self.llm.acall(prompt)
# 提取数字
import re
numbers = re.findall(r'0\.\d+|1\.0|1', response)
if numbers:
return float(numbers[0])
except:
pass
return 0.5 # 默认中等相关度
async def _generate_report(self, task: ResearchTask) -> str:
"""基于收集的资料生成研究报告"""
print("\n[生成报告] 正在整理研究成果...")
# 准备资料摘要
notes_summary = "\n\n".join([
f"[笔记 {i+1}] 来源: {note.source}\n{note.content[:500]}"
for i, note in enumerate(task.collected_notes[:10])
])
prompt = f"""
基于以下研究资料，生成一份结构化的研究报告。
研究主题: {task.topic}
子问题:
{chr(10).join(f"- {q}" for q in task.sub_questions)}
收集的资料:
{notes_summary}
请按以下结构生成报告:
1. 摘要
2. 研究背景
3. 主要发现 (按子问题组织)
4. 结论与建议
5. 参考资料
报告要求:
- 使用Markdown格式
- 内容准确、客观
- 引用来源
"""
report = await self.llm.acall(prompt)
return report
async def answer_question(self, question: str) -> str:
"""基于已有笔记回答问题"""
# 搜索相关笔记
relevant_notes = self.notes.search(question)
if not relevant_notes:
return "抱歉，现有资料中找不到相关信息。建议先进行相关主题的研究。"
# 构建上下文
context = "\n\n".join([
f"[来源: {note.source}]\n{note.content}"
for note in relevant_notes[:5]
])
prompt = f"""
基于以下资料，回答问题。
资料:
{context}
问题: {question}
请给出准确、简洁的回答，并注明信息来源。
"""
return await self.llm.acall(prompt)
# 使用示例
async def demo_research_agent():
"""演示研究助手的完整使用流程"""
# 初始化组件
search_tool = SearchTool(api_key="your-api-key")
note_storage = NoteStorage("my_research_notes.json")
# 这里用模拟的LLM客户端
class MockLLM:
async def acall(self, prompt: str) -> str:
# 模拟LLM响应
if "分解" in prompt:
return json.dumps([
"气候变化的科学原理是什么？",
"气候变化对农业生产的具体影响有哪些？",
"农业适应气候变化的策略有哪些？",
"未来的发展趋势和预测如何？"
])
elif "相关度" in prompt:
return "0.85"
elif "报告" in prompt:
return """# 研究报告：气候变化对农业的影响
## 1. 摘要
气候变化正在深刻影响全球农业生产...
## 2. 研究背景
农业是受气候变化影响最直接的行业之一...
## 3. 主要发现
### 3.1 气候变化的科学原理
温室气体排放导致全球气温上升...
### 3.2 对农业的具体影响
- 作物产量波动增加
- 病虫害范围扩大
- 水资源分布改变
## 4. 结论与建议
需要采取综合性的适应策略...
## 5. 参考资料
[1] IPCC报告
[2] FAO农业展望
"""
else:
return "这是基于资料的详细回答..."
llm = MockLLM()
# 创建研究助手
agent = ResearchAgent(llm, search_tool, note_storage)
# 执行研究
topic = "气候变化对农业的影响"
task = await agent.research(topic)
print("\n" + "="*60)
print("生成的研究报告:")
print("="*60)
print(task.report)
# 问答功能
print("\n" + "="*60)
print("问答测试:")
print("="*60)
answer = await agent.answer_question("气候变化如何影响水稻产量？")
print(f"Q: 气候变化如何影响水稻产量？")
print(f"A: {answer}")
# 运行演示
if __name__ == "__main__":
asyncio.run(demo_research_agent())8.2.5 工作流程图
研究助手工作流程
┌─────────────────────────────────────────────────────────────┐
│ │
│ 开始 │
│ │ │
│ ▼ │
│ 输入研究主题 │
│ │ │
│ ▼ │
│ LLM分解子问题 ──► 生成3-5个研究子问题 │
│ │ │
│ ▼ │
│ 对每个子问题: │
│ │ │
│ ├──► 搜索相关资料 │
│ │ │ │
│ │ ▼ │
│ │ 评估信息价值 ──► 相关度>0.6? ──否──► 跳过 │
│ │ │ │ │
│ │ │ 是 │
│ │ │ ▼ │
│ │ │ 获取详细内容 │
│ │ │ │ │
│ │ │ ▼ │
│ │ │ 保存到笔记库 │
│ │ │ │
│ ◄──────┘ │
│ │ (继续下一个子问题) │
│ ▼ │
│ 资料收集完成? ──否──► 继续搜索 │
│ │ │
│ 是 │
│ ▼ │
│ 基于笔记生成研究报告 │
│ │ │
│ ▼ │
│ 输出报告 + 支持问答 │
│ │
└─────────────────────────────────────────────────────────────┘
扩展建议：
- 接入真实的搜索引擎API（SerpAPI、Google Custom Search）
- 添加PDF/Word文档解析功能
- 实现知识图谱构建，增强关联分析能力
- 添加可视化界面（Gradio/Streamlit）
8.3 案例2：代码开发助手
8.3.1 需求分析
程序员在日常开发中需要：理解需求 → 设计架构 → 编写代码 → 测试 → 修复Bug。这个过程循环往复，耗费大量时间。一个智能的代码开发助手可以大幅提升开发效率。
目标：开发一个代码开发助手，能够理解自然语言需求，自动设计、编写、测试和修复代码，实现从需求到可运行代码的自动化。
8.3.2 核心功能
- 需求理解 - 解析自然语言需求，提取功能点
- 架构设计 - 设计代码结构和模块划分
- 代码生成 - 生成可运行的代码
- 自动测试 - 生成并执行测试用例
- Bug修复 - 分析错误并修复代码
8.3.3 系统架构
代码开发助手架构
┌─────────────────────────────────────────────────────────────────┐
│ │
│ 用户需求: "实现一个支持JWT认证的REST API登录系统" │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 规划层 (Planning) │ │
│ │ 需求分析 ──► 功能分解 ──► 技术选型 ──► 架构设计 │ │
│ └────────────────────────┬───────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 执行层 (Execution) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 代码生成 │ │ 代码执行 │ │ 测试运行 │ │ │
│ │ │ 工具 │ │ 工具 │ │ 工具 │ │ │
│ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │
│ └───────┼────────────┼────────────┼────────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 反思层 (Reflection) │ │
│ │ ┌────────────────────────────────────────────┐ │ │
│ │ │ 测试结果分析 │ │ │
│ │ │ ├─ 通过? ──► 完成 │ │ │
│ │ │ └─ 失败? ──► 诊断错误 ──► 修复 ──► 重试 │ │ │
│ │ └────────────────────────────────────────────┘ │ │
│ └────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
8.3.4 完整代码实现
"""
代码开发助手 - Code Development Agent
功能：理解需求、设计方案、编写代码、测试、修复Bug
架构：规划 + 工具调用 + 反思循环
"""
import os
import re
import json
import subprocess
import asyncio
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
class TaskStatus(Enum):
PENDING = "待处理"
PLANNING = "规划中"
CODING = "编码中"
TESTING = "测试中"
DEBUGGING = "调试中"
COMPLETED = "已完成"
FAILED = "失败"
@dataclass
class CodeFile:
"""代码文件"""
path: str
content: str
language: str
@dataclass
class TestResult:
"""测试结果"""
passed: bool
output: str
error: str = ""
coverage: float = 0.0
class CodeExecutor:
"""代码执行工具"""
def execute(self, code: str, language: str = "python") -> Tuple[bool, str]:
"""执行代码，返回(是否成功, 输出)"""
if language == "python":
return self._execute_python(code)
elif language == "javascript":
return self._execute_javascript(code)
else:
return False, f"不支持的语言: {language}"
def _execute_python(self, code: str) -> Tuple[bool, str]:
"""执行Python代码"""
try:
# 创建临时文件
with open("temp_code.py", "w", encoding="utf-8") as f:
f.write(code)
# 执行代码
result = subprocess.run(
["python", "temp_code.py"],
capture_output=True,
text=True,
timeout=10
)
success = result.returncode == 0
output = result.stdout if success else result.stderr
# 清理
os.remove("temp_code.py")
return success, output
except subprocess.TimeoutExpired:
return False, "执行超时"
except Exception as e:
return False, str(e)
def _execute_javascript(self, code: str) -> Tuple[bool, str]:
"""执行JavaScript代码（需要Node.js）"""
try:
with open("temp_code.js", "w", encoding="utf-8") as f:
f.write(code)
result = subprocess.run(
["node", "temp_code.js"],
capture_output=True,
text=True,
timeout=10
)
success = result.returncode == 0
output = result.stdout if success else result.stderr
os.remove("temp_code.js")
return success, output
except Exception as e:
return False, str(e)
class TestRunner:
"""测试框架"""
def run_tests(self, test_code: str, source_code: str) -> TestResult:
"""运行测试"""
try:
# 合并测试代码和源代码
full_code = source_code + "\n\n" + test_code
executor = CodeExecutor()
success, output = executor.execute(full_code)
return TestResult(
passed=success,
output=output,
error="" if success else output
)
except Exception as e:
return TestResult(
passed=False,
output="",
error=str(e)
)
def generate_tests(self, code: str, llm_client) -> str:
"""使用LLM生成测试代码"""
prompt = f"""
为以下代码生成pytest单元测试：
```python
{code}
```
要求：
1. 覆盖主要功能路径
2. 包含边界条件测试
3. 使用pytest框架
4. 只返回测试代码，不要解释
"""
return llm_client.call(prompt)
class CodeDevAgent:
"""代码开发智能体"""
def __init__(self, llm_client, max_retries: int = 3):
self.llm = llm_client
self.max_retries = max_retries
self.executor = CodeExecutor()
self.tester = TestRunner()
async def develop(self, requirement: str) -> Dict:
"""
完整的开发流程
"""
print(f"\n{'='*60}")
print(f"开始开发: {requirement}")
print(f"{'='*60}\n")
# 步骤1: 分析需求并设计
design = await self._analyze_and_design(requirement)
print("[设计完成]")
print(f" 功能列表: {design['features']}")
print(f" 技术栈: {design['tech_stack']}")
print(f" 文件结构: {design['file_structure']}")
# 步骤2: 生成代码
code_files = await self._generate_code(design)
print(f"\n[代码生成完成] 共 {len(code_files)} 个文件")
for f in code_files:
print(f" - {f.path}")
# 步骤3: 测试和迭代
results = []
for code_file in code_files:
result = await self._test_and_fix(code_file)
results.append(result)
# 汇总结果
all_passed = all(r['test_result'].passed for r in results)
return {
"requirement": requirement,
"design": design,
"code_files": code_files,
"test_results": results,
"success": all_passed
}
async def _analyze_and_design(self, requirement: str) -> Dict:
"""分析需求并生成设计方案"""
prompt = f"""
分析以下开发需求，生成详细的设计方案。
需求: {requirement}
请以JSON格式返回设计方案：
{{
"features": ["功能1", "功能2", ...], // 功能列表
"tech_stack": "Python Flask", // 技术选型
"file_structure": ["app.py", "models.py", ...], // 文件结构
"architecture": "MVC架构描述...", // 架构描述
"interfaces": ["API端点描述..."] // 接口定义
}}
只返回JSON，不要有其他内容。
"""
response = await self.llm.acall(prompt)
try:
design = json.loads(response)
except:
# 解析失败使用默认设计
design = {
"features": ["核心功能实现"],
"tech_stack": "Python",
"file_structure": ["main.py"],
"architecture": "简单模块化设计",
"interfaces": []
}
return design
async def _generate_code(self, design: Dict) -> List[CodeFile]:
"""根据设计生成代码"""
code_files = []
for file_info in design.get("file_structure", ["main.py"]):
prompt = f"""
根据以下设计方案，生成文件 {file_info} 的完整代码。
设计方案:
- 功能: {design.get('features')}
- 技术栈: {design.get('tech_stack')}
- 架构: {design.get('architecture')}
要求：
1. 代码完整可运行
2. 包含必要的注释
3. 遵循最佳实践
4. 只返回代码，不要解释
代码：
"""
code = await self.llm.acall(prompt)
# 提取代码块
code = self._extract_code(code)
# 检测语言
language = self._detect_language(file_info)
code_files.append(CodeFile(
path=file_info,
content=code,
language=language
))
return code_files
async def _test_and_fix(self, code_file: CodeFile) -> Dict:
"""测试并修复代码"""
print(f"\n[测试] {code_file.path}")
# 生成测试
test_code = self.tester.generate_tests(code_file.content, self.llm)
retry_count = 0
current_code = code_file.content
while retry_count < self.max_retries:
# 运行测试
test_result = self.tester.run_tests(test_code, current_code)
if test_result.passed:
print(f" ✓ 测试通过")
return {
"file": code_file.path,
"code": current_code,
"test_result": test_result,
"retry_count": retry_count
}
print(f" ✗ 测试失败 (第{retry_count + 1}次)")
print(f" 错误: {test_result.error[:100]}...")
# 修复代码
current_code = await self._fix_code(
current_code,
test_code,
test_result.error
)
retry_count += 1
print(f" ⚠ 达到最大重试次数，返回最后版本")
return {
"file": code_file.path,
"code": current_code,
"test_result": test_result,
"retry_count": retry_count
}
async def _fix_code(self, code: str, test_code: str,
error: str) -> str:
"""根据错误修复代码"""
prompt = f"""
以下代码在测试中出现了错误，请修复它。
原始代码:
```python
{code}
```
测试代码:
```python
{test_code}
```
错误信息:
{error}
请分析错误原因并修复代码。只返回修复后的完整代码。
"""
fixed_code = await self.llm.acall(prompt)
return self._extract_code(fixed_code)
def _extract_code(self, text: str) -> str:
"""从文本中提取代码块"""
# 尝试提取markdown代码块
patterns = [
r'```python\n(.*?)```',
r'```\n(.*?)```',
]
for pattern in patterns:
match = re.search(pattern, text, re.DOTALL)
if match:
return match.group(1).strip()
return text.strip()
def _detect_language(self, filename: str) -> str:
"""根据文件名检测编程语言"""
ext_map = {
'.py': 'python',
'.js': 'javascript',
'.ts': 'typescript',
'.java': 'java',
'.go': 'go',
'.rs': 'rust'
}
_, ext = os.path.splitext(filename)
return ext_map.get(ext, 'python')
# 使用示例：开发一个JWT认证系统
async def demo_code_dev():
"""演示代码开发助手的使用"""
class MockLLM:
"""模拟LLM客户端"""
async def acall(self, prompt: str) -> str:
if "设计方案" in prompt:
return json.dumps({
"features": [
"用户注册",
"用户登录",
"JWT Token生成",
"Token验证",
"受保护路由"
],
"tech_stack": "Python Flask + PyJWT",
"file_structure": ["app.py", "auth.py", "models.py"],
"architecture": "Flask REST API with JWT authentication",
"interfaces": [
"POST /register - 用户注册",
"POST /login - 用户登录",
"GET /protected - 受保护路由"
]
})
elif "app.py" in prompt:
return '''```python
from flask import Flask, request, jsonify
from auth import AuthManager
app = Flask(__name__)
auth = AuthManager()
@app.route('/register', methods=['POST'])
def register():
data = request.get_json()
username = data.get('username')
password = data.get('password')
if not username or not password:
return jsonify({"error": "缺少用户名或密码"}), 400
success = auth.register(username, password)
if success:
return jsonify({"message": "注册成功"}), 201
return jsonify({"error": "用户名已存在"}), 409
@app.route('/login', methods=['POST'])
def login():
data = request.get_json()
username = data.get('username')
password = data.get('password')
token = auth.login(username, password)
if token:
return jsonify({"token": token}), 200
return jsonify({"error": "登录失败"}), 401
@app.route('/protected', methods=['GET'])
def protected():
auth_header = request.headers.get('Authorization')
if not auth_header:
return jsonify({"error": "缺少认证信息"}), 401
token = auth_header.replace('Bearer ', '')
user = auth.verify_token(token)
if user:
return jsonify({"message": f"欢迎, {user}"}), 200
return jsonify({"error": "无效的token"}), 401
if __name__ == '__main__':
app.run(debug=True)
```'''
elif "auth.py" in prompt:
return '''```python
import jwt
import bcrypt
from datetime import datetime, timedelta
class AuthManager:
SECRET_KEY = "your-secret-key-change-in-production"
def __init__(self):
self.users = {} # 模拟用户数据库
def register(self, username: str, password: str) -> bool:
if username in self.users:
return False
# 密码哈希
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
self.users[username] = hashed
return True
def login(self, username: str, password: str) -> str:
if username not in self.users:
return None
stored_hash = self.users[username]
if bcrypt.checkpw(password.encode(), stored_hash):
# 生成JWT
payload = {
"username": username,
"exp": datetime.utcnow() + timedelta(hours=24)
}
return jwt.encode(payload, self.SECRET_KEY, algorithm="HS256")
return None
def verify_token(self, token: str) -> str:
try:
payload = jwt.decode(token, self.SECRET_KEY, algorithms=["HS256"])
return payload.get("username")
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
```'''
else:
return "```python\n# 代码内容\n```"
llm = MockLLM()
agent = CodeDevAgent(llm, max_retries=2)
# 执行开发任务
requirement = "实现一个支持JWT认证的REST API登录系统"
result = await agent.develop(requirement)
print("\n" + "="*60)
print("开发结果:")
print("="*60)
print(f"需求: {result['requirement']}")
print(f"成功: {'✓' if result['success'] else '✗'}")
print(f"\n生成的文件:")
for f in result['code_files']:
print(f"\n--- {f.path} ---")
print(f.content[:500] + "...")
if __name__ == "__main__":
asyncio.run(demo_code_dev())8.3.5 反思循环详解
反思循环 (Reflection Loop)
┌─────────────────────────────────────────────────────────────┐
│ │
│ ┌─────────────┐ │
│ │ 生成代码 │ │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ 执行测试 │ │
│ └──────┬──────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ │ │ │ │
│ ▼ │ ▼ │
│ ┌──────────┐ │ ┌──────────┐ │
│ │ 测试通过 │ │ │ 测试失败 │ │
│ │ ✓ │ │ │ ✗ │ │
│ └────┬─────┘ │ └────┬─────┘ │
│ │ │ │ │
│ ▼ │ ▼ │
│ ┌──────────┐ │ ┌──────────────┐ │
│ │ 完成交付 │ │ │ 分析错误 │ │
│ └──────────┘ │ └──────┬───────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ │ │ 定位问题 │ │
│ │ └──────┬───────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ │ │ 生成修复 │ │
│ │ └──────┬───────┘ │
│ │ │ │
│ └────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────┐ │
│ │ 重试次数 < 最大? │ │
│ │ 是 ────┐ │ │
│ │ 否 │ │ │
│ └─────────┬──┘ │ │
│ │ │ │
│ ┌─────────────┘ │ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 继续循环 │ │ 标记失败 │ │
│ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
8.4 案例3：智能客服系统
8.4.1 需求分析
传统客服面临压力大、响应慢、成本高的问题。智能客服系统可以7x24小时服务客户，自动解答常见问题，只在必要时转接人工。
目标：开发一个多智能体智能客服系统，能够理解客户问题、检索知识库、判断处理难度、生成回复或转人工。
8.4.2 核心功能
- 问题理解 - 分析客户问题意图和情感
- 知识检索 - 从知识库查找相关答案
- 智能分发 - 根据问题类型分配给合适的处理者
- 回复生成 - 生成自然、准确的回复
- 工单管理 - 复杂问题创建工单跟踪
8.4.3 多智能体架构
智能客服系统 - 多智能体架构
┌─────────────────────────────────────────────────────────────────┐
│ │
│ 客户输入 │
│ │ │
│ ▼ │
│ ┌────────────────────────┐ │
│ │ 接待员 Agent │ │
│ │ (Front Desk) │ │
│ │ • 意图识别 │ │
│ │ • 情感分析 │ │
│ │ • 问题分类 │ │
│ └───────────┬────────────┘ │
│ │ 分发 │
│ ┌───────────────┼───────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ 技术支持 │ │ 售后专员 │ │ 销售顾问 │ │
│ │ Agent │ │ Agent │ │ Agent │ │
│ │ │ │ │ │ │ │
│ │ 技术问题 │ │ 退换货 │ │ 产品咨询 │ │
│ │ Bug报告 │ │ 投诉处理 │ │ 报价服务 │ │
│ └─────┬──────┘ └─────┬──────┘ └─────┬──────┘ │
│ │ │ │ │
│ └───────────────┼───────────────┘ │
│ │ │
│ ▼ │
│ ┌────────────────────────┐ │
│ │ 质量检查 Agent │ │
│ │ (Quality Control) │ │
│ │ • 回复审核 │ │
│ │ • 满意度评估 │ │
│ └───────────┬────────────┘ │
│ │ │
│ ┌───────────┴───────────┐ │
│ │ │ │
│ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ │
│ │ 直接回复 │ │ 转人工 │ │
│ └────────────┘ └────────────┘ │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ 共享资源层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ 知识库 │ │ 对话历史 │ │ 工单系统 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
8.4.4 完整代码实现
"""
智能客服系统 - Multi-Agent Customer Service
功能：多智能体协作处理客户咨询
架构：主从模式 + 共享知识库
"""
import json
import asyncio
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
class IntentType(Enum):
TECH_SUPPORT = "技术支持"
AFTER_SALES = "售后服务"
SALES_INQUIRY = "销售咨询"
COMPLAINT = "投诉"
GENERAL = "一般咨询"
class Sentiment(Enum):
POSITIVE = "积极"
NEUTRAL = "中性"
NEGATIVE = "消极"
ANGRY = "愤怒"
@dataclass
class CustomerMessage:
"""客户消息"""
session_id: str
customer_id: str
content: str
timestamp: str
sentiment: Optional[Sentiment] = None
intent: Optional[IntentType] = None
@dataclass
class Ticket:
"""工单"""
id: str
session_id: str
type: str
description: str
status: str # open/in_progress/resolved/closed
assigned_to: Optional[str] = None
created_at: str = None
class KnowledgeBase:
"""知识库 - 客服系统的核心资产"""
def __init__(self):
self.faqs = {}
self.documents = []
self._load_data()
def _load_data(self):
"""加载知识库数据"""
# 模拟FAQ数据
self.faqs = {
"如何重置密码": "您可以在登录页面点击'忘记密码'，按照邮件指引重置。",
"支持哪些支付方式": "我们支持支付宝、微信支付、银行卡。",
"订单多久发货": "一般情况下，订单会在1-3个工作日内发货。",
"如何申请退款": "在订单页面点击'申请退款'，填写原因后提交。",
"产品保修期": "所有产品享受1年免费保修服务。"
}
def search(self, query: str, top_k: int = 3) -> List[Dict]:
"""搜索相关知识"""
results = []
query_lower = query.lower()
# 简单的关键词匹配（实际中使用向量检索）
for question, answer in self.faqs.items():
score = self._calculate_similarity(query_lower, question.lower())
if score > 0.3:
results.append({
"question": question,
"answer": answer,
"score": score
})
# 按相关度排序
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]
def _calculate_similarity(self, query: str, doc: str) -> float:
"""计算查询和文档的相似度"""
query_words = set(query.split())
doc_words = set(doc.split())
if not query_words:
return 0.0
overlap = len(query_words & doc_words)
return overlap / len(query_words)
class TicketSystem:
"""工单系统"""
def __init__(self):
self.tickets: Dict[str, Ticket] = {}
self.counter = 0
def create_ticket(self, session_id: str, ticket_type: str,
description: str) -> Ticket:
"""创建新工单"""
self.counter += 1
ticket_id = f"TK{self.counter:06d}"
ticket = Ticket(
id=ticket_id,
session_id=session_id,
type=ticket_type,
description=description,
status="open",
created_at=datetime.now().isoformat()
)
self.tickets[ticket_id] = ticket
print(f"[工单系统] 创建工单: {ticket_id}")
return ticket
def assign_ticket(self, ticket_id: str, agent_id: str):
"""分配工单"""
if ticket_id in self.tickets:
self.tickets[ticket_id].assigned_to = agent_id
self.tickets[ticket_id].status = "in_progress"
print(f"[工单系统] 工单 {ticket_id} 分配给 {agent_id}")
class MessageBus:
"""消息总线 - 智能体间通信"""
def __init__(self):
self.agents = {}
self.message_queue = asyncio.Queue()
def register(self, agent_id: str, agent):
self.agents[agent_id] = agent
async def send(self, sender: str, receiver: str,
message_type: str, content: Dict):
"""发送消息"""
msg = {
"sender": sender,
"receiver": receiver,
"type": message_type,
"content": content,
"timestamp": datetime.now().isoformat()
}
print(f"[{sender} -> {receiver}]: {message_type}")
if receiver in self.agents:
await self.agents[receiver].handle_message(msg)
class BaseAgent:
"""客服智能体基类"""
def __init__(self, agent_id: str, name: str, bus: MessageBus,
kb: KnowledgeBase, ticket_system: TicketSystem):
self.id = agent_id
self.name = name
self.bus = bus
self.kb = kb
self.tickets = ticket_system
async def handle_message(self, message: Dict):
"""处理接收到的消息"""
raise NotImplementedError
async def send_to(self, receiver: str, msg_type: str, content: Dict):
await self.bus.send(self.id, receiver, msg_type, content)
class FrontDeskAgent(BaseAgent):
"""接待员智能体 - 系统的入口"""
def __init__(self, bus: MessageBus, kb: KnowledgeBase,
ticket_system: TicketSystem, llm_client):
super().__init__("front_desk", "接待员", bus, kb, ticket_system)
self.llm = llm_client
self.session_history = {} # 会话历史
async def handle_customer(self, message: CustomerMessage):
"""处理客户消息"""
print(f"\n[{self.name}] 收到客户消息: {message.content}")
# 保存会话历史
if message.session_id not in self.session_history:
self.session_history[message.session_id] = []
self.session_history[message.session_id].append(message)
# 1. 分析意图和情感
intent, sentiment = await self._analyze_message(message.content)
message.intent = intent
message.sentiment = sentiment
print(f"[分析结果] 意图: {intent.value}, 情感: {sentiment.value}")
# 2. 根据意图分发
if sentiment == Sentiment.ANGRY:
# 愤怒客户直接转人工
await self._escalate_to_human(message, "客户情绪激动")
elif intent == IntentType.TECH_SUPPORT:
await self.send_to("tech_support", "new_case", {
"session_id": message.session_id,
"message": message.content,
"sentiment": sentiment.value
})
elif intent == IntentType.AFTER_SALES:
await self.send_to("after_sales", "new_case", {
"session_id": message.session_id,
"message": message.content,
"sentiment": sentiment.value
})
elif intent == IntentType.SALES_INQUIRY:
await self.send_to("sales", "new_case", {
"session_id": message.session_id,
"message": message.content
})
else:
# 一般咨询尝试直接回答
answer = await self._answer_from_kb(message.content)
if answer:
print(f"[{self.name}] 直接回答: {answer}")
else:
await self._escalate_to_human(message, "知识库无匹配")
async def _analyze_message(self, content: str) -> Tuple[IntentType, Sentiment]:
"""分析消息意图和情感"""
# 使用LLM进行分析（这里用简单规则模拟）
# 情感分析
angry_keywords = ["生气", "愤怒", "糟糕", "垃圾", "投诉", "垃圾", "失望"]
negative_keywords = ["不好", "问题", "错误", "失败", "慢"]
if any(kw in content for kw in angry_keywords):
sentiment = Sentiment.ANGRY
elif any(kw in content for kw in negative_keywords):
sentiment = Sentiment.NEGATIVE
else:
sentiment = Sentiment.NEUTRAL
# 意图识别
if any(kw in content for kw in ["bug", "错误", "崩溃", "无法", "报错"]):
intent = IntentType.TECH_SUPPORT
elif any(kw in content for kw in ["退款", "退货", "售后", "保修"]):
intent = IntentType.AFTER_SALES
elif any(kw in content for kw in ["价格", "多少钱", "优惠", "购买"]):
intent = IntentType.SALES_INQUIRY
else:
intent = IntentType.GENERAL
return intent, sentiment
async def _answer_from_kb(self, query: str) -> Optional[str]:
"""从知识库查找答案"""
results = self.kb.search(query)
if results and results[0]["score"] > 0.5:
return results[0]["answer"]
return None
async def _escalate_to_human(self, message: CustomerMessage, reason: str):
"""转接人工"""
print(f"[{self.name}] 转人工处理，原因: {reason}")
# 创建工单
ticket = self.tickets.create_ticket(
message.session_id,
"human_handover",
f"{reason}: {message.content}"
)
# 通知客户
print(f"系统回复: 您的问题已转接人工客服，工单号: {ticket.id}")
async def handle_message(self, message: Dict):
"""处理其他智能体的回复"""
if message["type"] == "response":
# 将回复传递给客户
content = message["content"]
print(f"\n[{self.name}] 转发给客户: {content['response']}")
# 质量检查
await self.send_to("quality_check", "check", {
"session_id": content["session_id"],
"response": content["response"]
})
class TechSupportAgent(BaseAgent):
"""技术支持智能体"""
def __init__(self, bus: MessageBus, kb: KnowledgeBase,
ticket_system: TicketSystem, llm_client):
super().__init__("tech_support", "技术支持", bus, kb, ticket_system)
self.llm = llm_client
async def handle_message(self, message: Dict):
if message["type"] == "new_case":
content = message["content"]
print(f"\n[{self.name}] 处理技术问题: {content['message']}")
# 1. 搜索知识库
kb_results = self.kb.search(content['message'])
# 2. 生成回复
response = await self._generate_response(
content['message'],
kb_results
)
# 3. 如果无法解决，创建工单
if "无法解决" in response or content.get('sentiment') == '愤怒':
ticket = self.tickets.create_ticket(
content['session_id'],
"tech_issue",
content['message']
)
self.tickets.assign_ticket(ticket.id, self.id)
response = f"{response}\n已为您创建工单: {ticket.id}，技术团队将跟进处理。"
# 4. 返回给接待员
await self.send_to("front_desk", "response", {
"session_id": content['session_id'],
"response": response
})
async def _generate_response(self, query: str,
kb_results: List[Dict]) -> str:
"""生成技术回复"""
if kb_results:
return f"根据常见问题，{kb_results[0]['answer']}"
# 模拟LLM生成回复
return f"我理解您遇到了'{query}'的问题。请先尝试重启应用，如果问题仍然存在，请提供具体的错误信息以便进一步诊断。"
class AfterSalesAgent(BaseAgent):
"""售后专员智能体"""
async def handle_message(self, message: Dict):
if message["type"] == "new_case":
content = message["content"]
print(f"\n[{self.name}] 处理售后问题: {content['message']}")
# 售后处理逻辑
response = "感谢您的反馈。关于售后问题，请您提供订单号，我们将尽快为您处理。" \
if "订单" in content['message'] else \
"我理解您的售后需求。请您提供更多细节，如购买时间、产品型号等，以便我们更好地为您服务。"
await self.send_to("front_desk", "response", {
"session_id": content['session_id'],
"response": response
})
class SalesAgent(BaseAgent):
"""销售顾问智能体"""
async def handle_message(self, message: Dict):
if message["type"] == "new_case":
content = message["content"]
print(f"\n[{self.name}] 处理销售咨询: {content['message']}")
# 销售咨询处理
response = "感谢您的咨询！我们的产品价格根据配置有所不同，基础版￥999起，专业版￥1999起。" \
"请问您主要关注哪些功能？我可以为您推荐最合适的方案。"
await self.send_to("front_desk", "response", {
"session_id": content['session_id'],
"response": response
})
class QualityCheckAgent(BaseAgent):
"""质量检查智能体"""
def __init__(self, bus: MessageBus, kb: KnowledgeBase,
ticket_system: TicketSystem):
super().__init__("quality_check", "质量检查", bus, kb, ticket_system)
self.check_count = 0
self.pass_count = 0
async def handle_message(self, message: Dict):
if message["type"] == "check":
self.check_count += 1
content = message["content"]
# 简单的质量检查
response = content['response']
issues = []
# 检查回复长度
if len(response) < 10:
issues.append("回复过短")
# 检查是否有礼貌用语
polite_words = ["请", "谢谢", "您好", "感谢"]
if not any(w in response for w in polite_words):
issues.append("缺少礼貌用语")
if not issues:
self.pass_count += 1
print(f"[{self.name}] 质量检查通过 ✓")
else:
print(f"[{self.name}] 质量检查未通过: {', '.join(issues)}")
# 记录质检统计
if self.check_count % 5 == 0:
print(f"\n[质检统计] 总检查: {self.check_count}, "
f"通过: {self.pass_count}, "
f"通过率: {self.pass_count/self.check_count*100:.1f}%")
# 系统演示
async def demo_customer_service():
"""演示智能客服系统"""
print("="*60)
print(" 智能客服系统启动")
print("="*60)
# 初始化基础设施
bus = MessageBus()
kb = KnowledgeBase()
ticket_system = TicketSystem()
# 模拟LLM
class MockLLM:
async def acall(self, prompt: str) -> str:
return "模拟LLM响应"
llm = MockLLM()
# 创建智能体
front_desk = FrontDeskAgent(bus, kb, ticket_system, llm)
tech_support = TechSupportAgent(bus, kb, ticket_system, llm)
after_sales = AfterSalesAgent(bus, kb, ticket_system)
sales = SalesAgent(bus, kb, ticket_system)
quality = QualityCheckAgent(bus, kb, ticket_system)
# 注册到消息总线
bus.register("front_desk", front_desk)
bus.register("tech_support", tech_support)
bus.register("after_sales", after_sales)
bus.register("sales", sales)
bus.register("quality_check", quality)
# 模拟客户对话
test_messages = [
("session_1", "user_001", "我的应用崩溃了，显示错误代码500"),
("session_2", "user_002", "如何重置密码？"),
("session_3", "user_003", "你们的产品多少钱？"),
("session_4", "user_004", "我要退货，太垃圾了！"),
("session_5", "user_005", "保修期是多久？"),
]
for session_id, user_id, content in test_messages:
print("\n" + "-"*60)
message = CustomerMessage(
session_id=session_id,
customer_id=user_id,
content=content,
timestamp=datetime.now().isoformat()
)
await front_desk.handle_customer(message)
await asyncio.sleep(0.5) # 模拟处理时间
print("\n" + "="*60)
print("客服系统演示结束")
print("="*60)
if __name__ == "__main__":
asyncio.run(demo_customer_service())8.4.5 工作流程图
客服处理流程
┌─────────────────────────────────────────────────────────────┐
│ │
│ 客户发送消息 │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ 接待员 Agent │ │
│ │ ┌─────────┐ ┌─────────┐ ┌────────┐ │ │
│ │ │意图识别 │ │情感分析 │ │分类 │ │ │
│ │ └────┬────┘ └────┬────┘ └───┬────┘ │ │
│ └───────┼──────────┼────────┼───────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────┐ │
│ │ 根据结果决策 │ │
│ │ │ │
│ │ 情感=愤怒? ──是──► 转人工 │ │
│ │ │ │ │
│ │ 否 │ │
│ │ ▼ │ │
│ │ 按意图分发到专业Agent │ │
│ └─────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ 专业 Agent 处理 │ │
│ │ (技术/售后/销售) │ │
│ │ • 检索知识库 │ │
│ │ • 生成回复 │ │
│ │ • 判断是否需要创建工单 │ │
│ └─────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────┐ │
│ │ 质量检查 Agent │ │
│ │ • 审核回复内容 │ │
│ │ • 检查礼貌用语 │ │
│ │ • 记录质检结果 │ │
│ └─────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 返回给客户 │
│ │
└─────────────────────────────────────────────────────────────┘
8.5 开发框架推荐
在实际开发中，可以使用成熟的框架来加速开发。以下是主流的智能体开发框架：
8.5.1 LangChain
LangChain 是目前最流行的LLM应用开发框架，提供了完整的智能体开发工具链。
核心特性：
核心特性：
- 丰富的预置工具和工具调用机制
- 支持多种LLM提供商（OpenAI、Anthropic、本地模型等）
- 内置ReAct、Plan-and-Execute等智能体类型
- 完整的记忆系统（Buffer、Vector、Entity等）
- Chain组合和LCEL表达式语言
# LangChain 快速示例
from langchain import OpenAI, LLMChain, PromptTemplate
from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.memory import ConversationBufferMemory
# 定义工具
tools = [
Tool(
name="Search",
func=search_function,
description="用于搜索信息的工具"
)
]
# 创建ReAct智能体
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, memory=memory)
# 运行
response = agent_executor.run("帮我查找今天的天气")8.5.2 AutoGPT
AutoGPT 是一个实验性的自主智能体框架，目标是让AI能够自主完成复杂任务。
核心特性：
核心特性：
- 自主目标分解和任务规划
- 自动化的工具调用循环
- 长期记忆和自我反思
- 支持多模态输入/输出
8.5.3 MetaGPT
MetaGPT 是一个多智能体协作框架，专注于软件开发场景。
核心特性：
核心特性：
- 预定义的专业角色（产品经理、架构师、工程师等）
- 标准化的软件开发流程（SOP）
- 结构化的通信协议
- 代码生成和项目管理能力
8.5.4 框架对比
| 框架 | 主要定位 | 学习曲线 | 适用场景 |
|---|---|---|---|
| LangChain | 通用LLM应用开发 | 中等 | 几乎所有智能体应用 |
| AutoGPT | 自主智能体 | 较高 | 自动化任务、研究 |
| MetaGPT | 多智能体软件开发 | 中等 | 自动化编程项目 |
| Microsoft AutoGen | 多智能体对话 | 中等 | 对话式AI应用 |
| CrewAI | 多智能体团队 | 较低 | 团队协作任务 |
8.6 学习路径总结
经过本教程的学习，你已经掌握了智能体开发的核心知识。以下是推荐的学习路径总结：
智能体开发学习路径
┌─────────────────────────────────────────────────────────────────┐
│ │
│ Level 1: 基础入门 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ✓ 理解智能体基本概念 │ │
│ │ ✓ 掌握ReAct模式 │ │
│ │ ✓ 学会工具调用 │ │
│ │ → 能开发简单的单智能体应用 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Level 2: 进阶提升 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ✓ 记忆系统设计 │ │
│ │ ✓ 提示词工程优化 │ │
│ │ ✓ 错误处理和鲁棒性 │ │
│ │ → 能开发复杂的单智能体应用 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Level 3: 系统掌握 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ✓ 多智能体协作模式 │ │
│ │ ✓ 角色分工与通信设计 │ │
│ │ ✓ 完整项目开发 │ │
│ │ → 能开发企业级多智能体系统 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ Level 4: 精通优化 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ✓ 性能优化与成本控制 │ │
│ │ ✓ 安全与隐私保护 │ │
│ │ ✓ 生产环境部署 │ │
│ │ → 能构建生产级智能体产品 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
进阶学习建议
- 动手实践 - 从改造第8章的案例开始，添加自己的功能
- 阅读源码 - 学习LangChain等框架的源码，理解实现细节
- 参与开源 - 为开源智能体项目贡献代码
- 关注前沿 - 跟踪最新的智能体研究论文和技术动态
8.7 资源推荐
8.7.1 开源项目
| 项目 | GitHub | 说明 |
|---|---|---|
| LangChain | langchain-ai/langchain | 最流行的LLM应用框架 |
| AutoGPT | Significant-Gravitas/AutoGPT | 自主智能体实验项目 |
| MetaGPT | geekan/MetaGPT | 多智能体软件开发框架 |
| AutoGen | microsoft/autogen | 微软多智能体对话框架 |
| CrewAI | joaomdmoura/crewAI | 多智能体团队框架 |
8.7.2 推荐论文
- ReAct - "ReAct: Synergizing Reasoning and Acting in Language Models"
- Chain-of-Thought - "Chain-of-Thought Prompting Elicits Reasoning in LLMs"
- Toolformer - "Toolformer: Language Models Can Teach Themselves to Use Tools"
- GPT-4 Technical Report - OpenAI的官方技术报告
8.7.3 社区与平台
- Hugging Face - 模型、数据集和Spaces应用平台
- GitHub Awesome-LLM-Agents - 智能体资源汇总
- Papers with Code - 最新研究论文和代码实现
- Discord社区 - LangChain、AutoGPT等官方Discord
课程总结
🎉 恭喜完成本教程！
通过这8章的学习，你已经掌握了智能体开发的核心技能：
- 第1章 - 了解AI智能体的基本概念和应用场景
- 第2章 - 掌握ReAct模式，让AI能推理和行动
- 第3章 - 学会为智能体配备各种工具
- 第4章 - 设计智能体的记忆系统
- 第5章 - 掌握提示词工程的核心技巧
- 第6章 - 学会调试、测试和部署智能体
- 第7章 - 理解多智能体协作的原理和模式
- 第8章 - 通过实战案例综合运用所学知识
下一步行动建议
- 完成一个个人项目 - 基于教程中的案例，开发一个解决你实际需求的智能体
- 加入社区 - 参与智能体开发者社区，学习他人经验
- 持续学习 - AI领域发展迅速，保持对新技术和新方法的关注
- 分享知识 - 将你的学习心得和项目经验分享给他人
📚 智能体开发入门教程 完
感谢你的学习！希望本教程能帮助你开启AI智能体开发的旅程。
有任何问题或建议，欢迎反馈交流。
🚀 Happy Coding! 🤖