1. 智能体开发基础概念解析
1.1 什么是智能体(Agent)
智能体是一种能够自主规划、决策和执行任务的软件组件。在人工智能领域,智能体的核心价值在于让大语言模型(LLM)具备了"做事"的能力,而不仅仅是"回答问题"。
传统的大语言模型存在明显的局限性:
- 只能基于训练数据回答问题
- 无法获取实时数据
- 难以执行复杂计算
- 缺乏调用外部工具的能力
智能体架构通过以下方式突破这些限制:
- 思考任务步骤:分析问题并制定解决方案
- 选择合适工具:根据需求调用最佳工具
- 执行工具调用:与外部系统交互
- 调整策略:根据结果优化后续行动
1.2 智能体的核心组成
一个完整的智能体通常包含三大核心组件:
- 大语言模型(大脑):负责推理、决策和自然语言处理
- 工具集(手脚):提供各种功能扩展,如数据查询、计算等
- 决策逻辑(思维):控制工作流程和策略调整
这种架构使得智能体能够从单纯的对话系统升级为可以实际影响现实世界的智能助手。
1.3 Chain与Agent的区别
虽然Chain和Agent都能增强大语言模型的能力,但它们的侧重点不同:
| 特性 | Chain | Agent |
|---|---|---|
| 主要功能 | 优化用户交互体验 | 扩展模型能力边界 |
| 实现手段 | 提示词模板、历史消息、RAG | 工具调用、自主决策 |
| 典型应用 | 连续对话、上下文理解 | 复杂任务执行、外部系统集成 |
| 输出形式 | 文本回复 | 行动结果+文本解释 |
以电商场景为例:
- Chain:优化商品问答的对话流畅度
- Agent:实现库存查询、订单处理等实际业务操作
2. 智能体开发实战
2.1 基础工具调用实现
智能体开发的核心是工具调用能力。我们通过一个天气查询示例来演示基础实现:
python复制from langchain.agents import create_agent
from langchain_community.chat_models import ChatTongyi
from langchain_core.tools import tool
@tool(description="查询天气")
def get_weather() -> str:
return "天气晴朗,温度23度"
agent = create_agent(
model=ChatTongyi(model="qwen3-max"),
tools=[get_weather],
system_prompt="你是一个聊天助手,可以回答用户问题"
)
response = agent.invoke({
"messages": [
{"role": "user", "content": "明天深圳天气怎么样?"}
]
})
print(response)
这个简单示例展示了:
- 使用
@tool装饰器定义工具 - 通过
create_agent创建智能体实例 - 将工具注册到智能体中
- 用户提问触发工具调用
2.2 流式输出处理
智能体支持流式输出,这对用户体验至关重要:
python复制for chunk in agent.stream(
{"messages": [{"role": "user", "content": "传智教育的股价是多少,并介绍一下"}]},
stream_mode="values"
):
latest_message = chunk['messages'][-1]
if latest_message.content:
print(latest_message.content)
if hasattr(latest_message, 'tool_calls'):
print(f"工具调用:{[tc['name'] for tc in latest_message.tool_calls]}")
流式输出的优势:
- 实时显示处理进度
- 降低用户等待焦虑
- 支持复杂任务的渐进式展示
2.3 ReAct框架实现
ReAct(Reasoning+Acting)是智能体的核心工作范式:
python复制agent = create_agent(
model=ChatTongyi(model="qwen3-max"),
tools=[get_weight, get_height],
system_prompt="""
你是严格遵循ReAct框架的智能体,必须按【思考->行动->观察->再思考】的流程解决问题,
且每轮仅能思考并调用1个工具。
并告知我你的思考过程,工具调用的原因。
"""
)
ReAct工作流程:
- 思考:分析问题并制定策略
- 行动:选择合适的工具执行
- 观察:评估工具返回结果
- 调整:根据结果优化后续行动
这种模式特别适合需要多步推理的复杂场景,如:
- 智能客服问题解决
- 数据分析报告生成
- 自动化任务规划
3. 高级功能开发
3.1 中间件开发
中间件可以对智能体的执行过程进行监控和干预:
python复制from langchain.agents.middleware import before_agent, after_agent
@before_agent
def log_before_agent(state: AgentState, runtime: Runnable) -> None:
print(f"Agent启动,附带{len(state['messages'])}条消息")
@after_agent
def log_after_agent(state: AgentState, runtime: Runnable) -> None:
print(f"Agent结束,附带{len(state['messages'])}条消息")
常见中间件应用场景:
- 日志记录与分析
- 性能监控
- 安全防护
- 异常处理
- 提示词动态调整
3.2 工具开发最佳实践
开发高质量工具需要注意以下要点:
- 明确的描述:
description参数要准确说明工具功能 - 类型注解:使用Python类型注解明确输入输出
- 错误处理:工具内部要有完善的异常捕获
- 性能优化:避免长时间阻塞操作
- 文档注释:详细说明使用方法和注意事项
示例工具开发:
python复制@tool(description="查询股票信息,传入股票名称,返回字符串信息")
def get_stock_info(name: str) -> str:
"""
获取股票基本信息
Args:
name: 股票名称(中文)
Returns:
股票基本信息字符串
Raises:
ValueError: 当股票不存在时抛出
"""
try:
# 模拟数据查询
if name == "传智教育":
return "传智教育(003032),A股上市公司,专注于IT职业教育"
raise ValueError("股票不存在")
except Exception as e:
return f"查询失败: {str(e)}"
3.3 智能体服务化
将智能体封装为可复用的服务:
python复制from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Query(BaseModel):
content: str
@app.post("/ask")
async def ask_agent(query: Query):
agent = ReactAgent()
response = []
for chunk in agent.execute_stream(query.content):
response.append(chunk)
return {"response": "".join(response)}
服务化带来的优势:
- 便于系统集成
- 支持多客户端访问
- 方便扩展和负载均衡
- 简化权限管理
4. 实战项目:智扫通智能客服
4.1 项目架构设计
智扫通是一个面向消费者的扫地机器人智能客服系统,主要架构如下:
code复制├── agent/ # 智能体核心
│ ├── tools/ # 工具集合
│ └── middleware/ # 中间件
├── rag/ # 检索增强模块
│ ├── vector_store/ # 向量存储
│ └── service.py # RAG服务
├── web/ # 前端界面
│ └── streamlit/ # Streamlit实现
└── utils/ # 工具函数
4.2 核心工具开发
项目包含多种实用工具:
- 天气查询工具:
python复制@tool(description="获取指定城市的天气信息")
def get_weather(city: str) -> str:
# 实际项目中这里会调用天气API
return f"{city}天气晴朗,温度26℃"
- 用户信息工具:
python复制@tool(description="获取用户所在城市")
def get_user_location() -> str:
# 模拟从用户系统获取数据
return random.choice(["北京", "上海", "广州", "深圳"])
- RAG检索工具:
python复制@tool(description="从知识库检索相关信息")
def rag_search(query: str) -> str:
return rag_service.search(query)
4.3 流式输出实现
使用Streamlit实现流畅的用户界面:
python复制import streamlit as st
# 初始化会话状态
if "messages" not in st.session_state:
st.session_state.messages = []
# 显示历史消息
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# 处理用户输入
if prompt := st.chat_input("请输入您的问题"):
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("assistant"):
response = st.write_stream(agent.execute_stream(prompt))
st.session_state.messages.append({"role": "assistant", "content": response})
4.4 性能优化技巧
- 工具调用缓存:
python复制from functools import lru_cache
@lru_cache(maxsize=100)
@tool(description="查询天气")
def get_weather(city: str) -> str:
# 实现代码
- 异步工具调用:
python复制import asyncio
@tool(description="并发查询多个信息")
async def query_multi_info(queries: list[str]) -> list[str]:
tasks = [query_single(q) for q in queries]
return await asyncio.gather(*tasks)
- 负载监控中间件:
python复制@wrap_tool_call
def monitor_tool_load(request, handler):
start_time = time.time()
try:
result = handler(request)
duration = time.time() - start_time
logger.info(f"工具{request.tool_call['name']}执行耗时: {duration:.2f}s")
return result
except Exception as e:
logger.error(f"工具执行异常: {str(e)}")
raise
5. 常见问题与解决方案
5.1 工具调用失败处理
问题现象:
- 工具执行超时
- 返回结果格式不符合预期
- 依赖服务不可用
解决方案:
- 实现重试机制:
python复制from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
@tool(description="查询外部数据")
def query_external_data():
# 容易失败的操作
- 添加备用方案:
python复制@tool(description="查询天气")
def get_weather(city: str) -> str:
try:
return weather_api.query(city)
except Exception:
return f"无法获取{city}实时天气,最后记录为晴转多云"
- 输入验证:
python复制from pydantic import validate_arguments
@validate_arguments
@tool(description="计算BMI")
def calculate_bmi(height: float, weight: float) -> float:
return weight / (height ** 2)
5.2 智能体决策优化
常见问题:
- 工具选择不合理
- 多步任务规划效率低
- 复杂场景处理能力不足
优化策略:
- 改进系统提示词:
python复制system_prompt = """
你是一个专业的扫地机器人客服助手,请遵循以下原则:
1. 首先确定用户问题的核心需求
2. 优先使用知识库检索(RAG)回答问题
3. 需要实时数据时才调用工具
4. 多步任务要分解并确认中间结果
"""
- 工具优先级设置:
python复制tools = [
{"tool": rag_search, "weight": 0.9},
{"tool": get_weather, "weight": 0.7},
# 其他工具...
]
- 决策过程记录与分析:
python复制@before_model
def log_decision_process(state, runtime):
logger.info(f"决策上下文: {state['messages'][-3:]}")
5.3 性能瓶颈排查
典型性能问题:
- 工具调用延迟高
- 大语言模型响应慢
- 复杂任务处理时间长
排查方法:
- 添加详细日志:
python复制import time
@wrap_tool_call
def time_tool_execution(request, handler):
start = time.time()
result = handler(request)
logger.info(f"工具{request.tool_call['name']}耗时: {time.time()-start:.2f}s")
return result
- 关键指标监控:
python复制# 监控智能体各阶段耗时
agent_duration = {
"thinking": 0.0,
"tool_calling": 0.0,
"model_inference": 0.0
}
- 负载测试:
python复制# 使用Locust等工具模拟并发请求
from locust import HttpUser, task
class AgentUser(HttpUser):
@task
def ask_question(self):
self.client.post("/ask", json={"content": "如何清洁扫地机器人?"})
5.4 安全防护措施
安全隐患:
- 恶意工具调用
- 敏感信息泄露
- 提示词注入攻击
防护方案:
- 输入过滤:
python复制@wrap_tool_call
def filter_malicious_input(request, handler):
if "DROP TABLE" in str(request.tool_call['args']):
raise ValueError("检测到恶意输入")
return handler(request)
- 权限控制:
python复制def create_agent_with_auth(user):
tools = [basic_tools]
if user.is_premium:
tools.extend(premium_tools)
return create_agent(tools=tools)
- 敏感数据脱敏:
python复制from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
def anonymize_text(text):
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
results = analyzer.analyze(text=text, language="en")
return anonymizer.anonymize(text=text, analyzer_results=results)
