基于 LangGraph 和 DeepSeek 的智能代理系统，支持 MCP 工具调用、任务分解和实时可视化
🆕 新增功能:
- ✨ 工作流可视化 - 实时展示智能体的思考过程和工具调用
- 🎯 智能旅游规划对话 - 多轮对话、信息收集、任务拆解、工具容错
AgenticAI 是一个功能完整的智能代理系统，能够理解复杂任务、自动分解执行、调用多种工具，并提供实时可视化界面。
- 智能理解 - 深度解析用户模糊或复杂的需求
- 自动分解 - 将复杂任务拆解为可执行的子任务
- 工具调度 - 智能选择并调用 25+ 种 MCP 工具
- 错误处理 - 多层兜底策略，保证任务完成率
- 🆕 工作流可视化 - 实时展示智能体的完整工作过程
- 🆕 MCP工具追踪 - 详细展示每个工具的调用和结果
- 多轮对话 - 支持信息收集和会话历史管理
- 智能容错 - MCP工具失败时自动降级到搜索兜底
后端:
- FastAPI - 高性能 Web 框架
- LangChain/LangGraph - 智能代理编排框架
- DeepSeek API - 大语言模型
- langchain-mcp-adapters - MCP 工具集成
前端:
- React 18 - UI 框架
- Axios - HTTP 客户端
- WebSocket - 实时通信
- Python 3.9+
- Node.js 16+
- DeepSeek API Key
- Conda (推荐使用langchain_stu环境)
cd backend
cp .env.example .env编辑 .env 文件：
# DeepSeek API 配置 (必需)
DEEPSEEK_API_KEY=your_deepseek_api_key_here
DEEPSEEK_BASE_URL=https://api.deepseek.com
# 服务配置 (可选)
BACKEND_PORT=8000
FRONTEND_PORT=3000chmod +x start.sh
./start.sh启动脚本会自动完成：
- ✅ 安装后端 Python 依赖
- ✅ 安装前端 Node.js 依赖
- ✅ 启动后端服务 (http://localhost:8000)
- ✅ 启动前端服务 (http://localhost:3000)
- 前端界面: http://localhost:3000
- API 文档: http://localhost:8000/docs
- 健康检查: http://localhost:8000/api/v2/health
后端 (使用Conda环境):
conda activate langchain_stu
cd backend
python -m uvicorn app.main:app --reload --port 8000前端:
cd frontend
npm install
npm start实时展示智能体（Agent）和MCP工具的完整工作过程！
-
智能体工作流程时间线
- 🔍 分析阶段 - 理解用户意图
- 🎯 规划阶段 - 任务拆解
- ⚡ 执行阶段 - 工具调用
- 📊 总结阶段 - 结果整合
-
MCP工具调用详情
- 工具名称和服务器
- 调用参数（JSON格式）
- 执行结果展示
- 性能数据统计
-
实时更新
- WebSocket实时推送
- 动画效果流畅
- 状态清晰标识
# 启动应用
./start.sh
# 访问前端
open http://localhost:3000
# 发送测试消息
"我想从北京去贵阳旅游"
# 观察工作流程可视化！效果预览：
🔍 分析阶段
├─ 识别关键词: 北京、贵阳、旅游
└─ 状态: ✅ 成功
🎯 规划阶段
├─ 拆解任务: 交通、景点、天气
└─ 状态: ✅ 成功
⚡ 执行阶段
├─ 🚄 12306__query_trains
│ ├─ 参数: {from: "北京", to: "贵阳"}
│ └─ 结果: 查询到12趟车次
├─ 🗺️ amap-web__search
│ └─ 结果: 15个景点
└─ 状态: ✅ 成功
📊 总结阶段
└─ 生成完整旅游方案
新增智能旅游规划对话系统,支持:
- ✅ 智能信息收集 - 自动识别缺失信息并友好追问
- ✅ 任务智能拆解 - 自动分解为时间计算、交通查询、景点推荐等子任务
- ✅ MCP工具容错 - 主工具失败时自动降级到搜索工具兜底
- ✅ 多轮对话 - 支持会话历史,可追加需求
方法1: 使用测试脚本
cd /nlp/DQ/AgenticAI
python test_travel_planning.py方法2: 使用Shell脚本
./test_travel_api.sh方法3: curl命令
# 第一轮: 信息不完整
curl -X POST http://localhost:8000/api/v2/tasks \
-H "Content-Type: application/json" \
-d '{"description": "我要去贵阳旅游"}'
# 第二轮: 补充信息 (使用返回的session_id)
curl -X POST http://localhost:8000/api/v2/tasks \
-H "Content-Type: application/json" \
-d '{
"description": "从北京出发坐高铁,2个人,玩5天,下周一出发",
"session_id": "上一步返回的session_id"
}'用户: "我要去贵阳旅游"
AI: "好的,我来帮你规划贵阳之旅!为了给你更好的建议,还需要确认几个信息:
1. 从哪里出发?
2. 打算坐高铁还是飞机?
3. 几个人去?
4. 计划玩几天?
5. 什么时候出发?"
状态: WAITING (等待用户补充信息)
---
用户: "从北京出发坐高铁,2个人,玩5天,下周一出发"
AI: [自动执行以下步骤]
1. ✅ 调用 time 工具计算"下周一"的具体日期
2. ✅ 调用 12306 查询北京到贵阳的高铁
- 如果失败 → 自动降级到 search 搜索"北京到贵阳高铁"
3. ✅ 调用 zhipu-web-search-sse 搜索贵阳旅游攻略
4. ✅ 调用 weather 查询贵阳天气
5. ✅ 调用 amap-web 查看景点地图位置
[返回完整的旅游规划方案]
状态: SUCCESS
- 完整指南: TRAVEL_PLANNING_GUIDE.md
- 快速测试: QUICK_TEST.md
系统能够深度解析用户的自然语言输入，自动拆解为可执行的子任务。
示例:
用户: "我想下周一从北京去上海，帮我查查交通和天气"
系统自动分解:
1. 📅 计算下周一的日期
2. ✈️ 查询北京到上海的航班信息
3. 🚄 查询北京到上海的火车票
4. 🌤️ 查询上海下周一的天气
5. 📊 汇总并推荐最佳方案
支持丰富的 MCP (Model Context Protocol) 工具集：
| 分类 | 工具 | 功能 |
|---|---|---|
| 🚗 交通出行 | 飞常准、12306 | 航班查询、火车票查询 |
| 🌤️ 生活服务 | 天气、菜谱、餐饮助手 | 天气预报、做菜指南、膳食规划 |
| 🗺️ 位置服务 | 高德地图 | 位置搜索、路线规划、周边查询 |
| 🔍 信息检索 | 通晓、维基百科 | 实时搜索、知识查询 |
| 🎨 AI 创作 | 魔搭文生图 | FLUX、Qwen-Image 模型 |
| 🌐 数据获取 | Fetch、超浏览器 | 网页抓取、内容提取 |
| 📈 资讯热点 | 趋势聚合 | 微博、知乎、B站热搜 |
| 🧠 AI 能力 | 思考工具 | 结构化推理、问题分析 |
多层兜底策略保证任务完成：
工具调用失败 → 尝试替代工具
↓ 失败
所有工具失败 → 启用大模型推理
↓ 失败
参数错误 → 自动分析并修正
↓ 失败
超时 → 自动重试
日志示例:
⚠️ 工具调用失败: weather__get_weather (超时)
🔄 尝试替代工具: weather__forecast
✅ 使用替代方案成功，返回结果
前端界面实时展示：
- ⏱️ 执行时间线 - 可视化任务执行过程
- 📋 子任务追踪 - 每个子任务的状态和结果
- 🔧 工具调用详情 - 显示调用的工具和参数
⚠️ 错误和重试 - 透明展示错误处理过程- 📊 结果汇总 - 结构化的最终结果展示
支持多轮对话式的需求澄清：
用户: "帮我查火车票"
系统: "好的，请问您要从哪里到哪里？什么时候出发？"
用户: "明天从北京到上海"
系统: "正在查询明天北京到上海的火车票..."
┌──────────────────────────────────────────────────────────┐
│ 前端 (React) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 任务创建 │ │ 执行监控 │ │ 结果展示 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└──────────────────────────────────────────────────────────┘
│ HTTP/WebSocket
┌──────────────────────────────────────────────────────────┐
│ 后端 (FastAPI) │
│ ┌───────────────────────────────────────────────────┐ │
│ │ SimpleMCPEngine (核心引擎) │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ LangChain Agent (自动工具调用) │ │ │
│ │ │ • 理解用户意图 │ │ │
│ │ │ • 选择合适工具 │ │ │
│ │ │ • 构造调用参数 │ │ │
│ │ │ • 处理工具结果 │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ └───────────────────────────────────────────────────┘ │
│ │ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ MCP Client (工具管理器) │ │
│ │ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │ │
│ │ │航班 │ │火车 │ │天气 │ │地图 │ │搜索 │...│ │
│ │ └──────┘ └──────┘ └──────┘ └──────┘ └──────┘ │ │
│ └───────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────┘
│
┌──────────────────────────────────────────────────────────┐
│ 外部 MCP 服务器 (ModelScope) │
│ 飞常准 | 12306 | 天气 | 高德 | 通晓 | ... │
└──────────────────────────────────────────────────────────┘
采用 LangChain 推荐的方式，让 Agent 自动管理工具调用：
class SimpleMCPEngine:
"""简化的 MCP 引擎 - 使用 create_agent 自动管理"""
async def initialize(self):
# 获取所有 MCP 工具
mcp_tools = await self.mcp_client._mcp_client.get_tools()
# 创建 Agent，传入工具
self.agent = create_agent(
model=self.llm,
tools=mcp_tools,
system_prompt="你是一个智能助手"
)
async def execute_task(self, task_id, user_input):
# Agent 自动完成所有步骤
result = await self.agent.ainvoke({
"messages": [HumanMessage(content=user_input)]
})
return result为什么这样设计？
- ✅ 简单 - 只需 50 行代码
- ✅ 可靠 - 利用 LangChain 成熟的工具调用机制
- ✅ 自动化 - Agent 自己选择工具、构造参数、处理结果
- ✅ 可扩展 - 添加新工具无需修改代码
管理所有 MCP 服务器的连接和配置：
class MCPClient:
MCP_SERVERS = {
"weather": {
"type": "sse",
"url": "https://mcp.api-inference.modelscope.net/.../sse",
"headers": {"Authorization": "Bearer ms-xxx"},
"description": "全球天气查询工具",
"keywords": ["天气", "气温", "weather"]
},
# ... 25+ 个工具
}
async def initialize(self):
# 使用 langchain-mcp-adapters 连接所有服务器
self._mcp_client = MultiServerMCPClient(servers_config)工具名称格式:
服务器名__工具名
- ✅
weather__get_weather - ✅
12306__query_trains - ❌
weather
提供 RESTful API 和 WebSocket 接口：
@router.post("/api/v2/tasks")
async def create_task(request: TaskRequest):
"""创建并执行任务"""
engine = get_engine()
result = await engine.execute_task(
task_id=task_id,
user_input=request.description,
thread_id=session_id
)
return TaskResponse(**result)
@router.websocket("/api/v2/ws/chat/{session_id}")
async def websocket_chat(websocket: WebSocket, session_id: str):
"""WebSocket 实时聊天"""
# 实时推送执行进度和结果所有 MCP 工具在 backend/app/mcp/client.py 中配置：
MCP_SERVERS = {
"weather": {
"type": "sse",
"url": "https://mcp.api-inference.modelscope.net/4faa12d916bc4a/sse",
"headers": {
"Authorization": "Bearer ms-52970502-e594-4344-b926-2b3bb2322242"
},
"description": "全球天气查询工具",
"keywords": ["天气", "气温", "下雨", "weather"],
"category": "生活服务"
},
# ... 更多工具配置
}- 在
MCP_SERVERS中添加配置
"new_tool": {
"type": "sse",
"url": "https://your-mcp-server-url/sse",
"headers": {"Authorization": "Bearer your-token"},
"description": "工具描述",
"keywords": ["关键词1", "关键词2"],
"category": "分类"
}- 重启服务
./start.sh- 验证工具加载
访问 http://localhost:8000/docs 查看工具列表，或查看日志：
✅ 加载了 26 个 MCP 工具
graph LR
A[用户输入] --> B[Agent 分析]
B --> C{需要工具?}
C -->|是| D[选择工具]
C -->|否| E[直接回复]
D --> F[构造参数]
F --> G[调用 MCP 工具]
G --> H{成功?}
H -->|是| I[处理结果]
H -->|否| J[重试/替代方案]
I --> K[生成回复]
J --> G
K --> L[返回用户]
AgenticAI/
├── backend/ # 后端服务
│ ├── app/
│ │ ├── main.py # FastAPI 入口
│ │ ├── api/
│ │ │ └── langgraph_routes.py # API 路由
│ │ ├── core/
│ │ │ └── config.py # 配置管理
│ │ ├── mcp/
│ │ │ └── client.py # MCP 客户端
│ │ ├── models/
│ │ │ └── schemas.py # 数据模型
│ │ └── services/
│ │ ├── langgraph_engine_fixed.py # ✨ 核心引擎
│ │ └── llm_client.py # LLM 客户端
│ ├── .env.example # 环境变量模板
│ └── requirements.txt # Python 依赖
├── frontend/ # 前端应用
│ ├── src/
│ │ ├── App.jsx # React 主应用
│ │ ├── pages/
│ │ │ ├── Chat.jsx # 聊天页面
│ │ │ └── TaskDetail.jsx # 任务详情
│ │ ├── components/
│ │ │ ├── ExecutionTimeline.jsx # 执行时间线
│ │ │ └── TaskSummary.jsx # 任务摘要
│ │ └── services/
│ │ └── api.js # API 封装
│ └── package.json # Node 依赖
├── docker-compose.yml # Docker 配置
├── start.sh # 🚀 启动脚本
└── README.md # 📖 本文档
1. 安装依赖
# 后端
cd backend
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
pip install -r requirements.txt
# 前端
cd ../frontend
npm install2. 配置环境变量
cp backend/.env.example backend/.env
# 编辑 .env，填入 DEEPSEEK_API_KEY3. 启动开发服务器
# 方式 1: 使用启动脚本 (推荐)
./start.sh
# 方式 2: 手动启动
# 终端 1 - 后端
cd backend
python -m uvicorn app.main:app --reload --port 8000
# 终端 2 - 前端
cd frontend
npm start创建任务:
curl -X POST http://localhost:8000/api/v2/tasks \
-H "Content-Type: application/json" \
-d '{
"description": "查询明天北京到上海的火车票",
"session_id": "test_session"
}'响应:
{
"task_id": "uuid",
"status": "SUCCESS",
"message": "查询到以下车次：...",
"session_id": "test_session"
}WebSocket 连接:
const ws = new WebSocket('ws://localhost:8000/api/v2/ws/chat/session_123');
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log('任务进度:', data);
};
ws.send(JSON.stringify({
message: "查询北京天气"
}));Python (后端):
- 遵循 PEP 8
- 使用类型提示
- 文档字符串使用 Google 风格
async def execute_task(
self,
task_id: str,
user_input: str,
thread_id: str = "default"
) -> Dict[str, Any]:
"""执行任务
Args:
task_id: 任务 ID
user_input: 用户输入
thread_id: 会话 ID
Returns:
任务执行结果字典
"""
# ...JavaScript (前端):
- 使用 ES6+
- 组件使用函数式 + Hooks
- 遵循 Airbnb 风格指南
const TaskDetail = ({ taskId }) => {
const [task, setTask] = useState(null);
useEffect(() => {
fetchTask(taskId).then(setTask);
}, [taskId]);
return <div>{/* ... */}</div>;
};1. 构建镜像
docker-compose build2. 启动服务
docker-compose up -d3. 查看日志
docker-compose logs -f backend
docker-compose logs -f frontend4. 停止服务
docker-compose down后端 (使用 Gunicorn):
gunicorn app.main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 120 \
--access-logfile - \
--error-logfile -前端 (构建静态文件):
cd frontend
npm run build
# 使用 Nginx 托管
server {
listen 80;
server_name your-domain.com;
root /path/to/frontend/build;
index index.html;
location / {
try_files $uri $uri/ /index.html;
}
location /api {
proxy_pass http://localhost:8000;
proxy_set_header Host $host;
}
}必需变量:
DEEPSEEK_API_KEY=sk-xxxxx # DeepSeek API Key可选变量:
DEEPSEEK_BASE_URL=https://api.deepseek.com
BACKEND_PORT=8000
FRONTEND_PORT=3000
LOG_LEVEL=INFO
CORS_ORIGINS=["http://localhost:3000"]1. 工具缓存
class SimpleMCPEngine:
def __init__(self):
self._agent_cache = None
async def initialize(self):
if self._agent_cache is None:
mcp_tools = await self.mcp_client._mcp_client.get_tools()
self._agent_cache = create_agent(
model=self.llm,
tools=mcp_tools
)2. 并发控制
from asyncio import Semaphore
class MCPClient:
def __init__(self):
self._semaphore = Semaphore(5) # 限制 5 个并发
async def call_tool(self, *args, **kwargs):
async with self._semaphore:
return await self._execute_tool(*args, **kwargs)3. 连接池
# 在 main.py 中
@app.on_event("startup")
async def startup():
# 预初始化 MCP 连接
engine = get_engine()
await engine.initialize()症状:
❌ 未找到工具: weather
❌ API 调用失败: INVALID_PARAMS
原因:
- 使用了旧版引擎
- 工具名称格式错误
解决:
确认 app/api/langgraph_routes.py 使用了正确的引擎：
from app.services.langgraph_engine_fixed import SimpleMCPEngine as LangGraphAgentEngine而不是：
from app.services.langgraph_engine import LangGraphAgentEngine # ❌ 旧版症状:
❌ Request timeout after 60 seconds
解决:
在 langgraph_engine_fixed.py 中增加超时时间：
self.llm = ChatOpenAI(
model="deepseek-chat",
api_key=deepseek_api_key,
base_url=deepseek_base_url,
temperature=0.7,
timeout=120 # 增加到 120 秒
)检查步骤:
- 确认后端运行
curl http://localhost:8000/docs- 检查 CORS 配置
在 app/main.py 中：
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境应设置具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)- 检查防火墙
# 确保端口 8000 和 3000 可访问
sudo ufw allow 8000
sudo ufw allow 3000Python 依赖:
# 清理并重新安装
pip install --upgrade pip
pip cache purge
pip install -r requirements.txt --force-reinstallNode.js 依赖:
# 清理并重新安装
rm -rf node_modules package-lock.json
npm cache clean --force
npm install后端日志:
# 实时查看
tail -f backend/app.log
# 搜索错误
grep "ERROR\|❌" backend/app.log | tail -20
# 查看工具调用
grep "🔧\|调用工具" backend/app.logDocker 日志:
docker-compose logs -f backend
docker-compose logs -f frontend
# 查看特定服务的最近 100 行
docker-compose logs --tail=100 backend1. 启用调试模式
# .env
LOG_LEVEL=DEBUG2. 测试 MCP 工具
# 在 Python 控制台
import asyncio
from app.mcp.client import get_mcp_client
async def test():
client = get_mcp_client()
await client.initialize()
tools = await client._mcp_client.get_tools()
print(f"加载了 {len(tools)} 个工具")
for tool in tools[:5]:
print(f" - {tool.name}")
asyncio.run(test())3. API 性能测试
# 使用 ab (Apache Bench)
ab -n 100 -c 10 -p request.json -T application/json \
http://localhost:8000/api/v2/tasks
# request.json 内容:
# {"description": "测试任务", "session_id": "test"}# ✅ 使用环境变量
export DEEPSEEK_API_KEY=sk-xxxxx
# ❌ 不要硬编码在代码中
api_key = "sk-xxxxx" # 危险！
# ❌ 不要提交到 Git
git add .env # 危险！确保 .env 在 .gitignore 中from pydantic import BaseModel, validator
class TaskRequest(BaseModel):
description: str
session_id: Optional[str] = None
@validator('description')
def validate_description(cls, v):
if len(v) > 1000:
raise ValueError('描述过长')
# 过滤敏感词汇
forbidden = ['SQL', 'DROP', 'DELETE']
for word in forbidden:
if word in v.upper():
raise ValueError(f'禁止使用: {word}')
return v# 添加 API Key 验证
from fastapi import Header, HTTPException
async def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != os.getenv("API_KEY"):
raise HTTPException(status_code=401, detail="Invalid API Key")
@router.post("/api/v2/tasks", dependencies=[Depends(verify_api_key)])
async def create_task(request: TaskRequest):
# ...from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@router.post("/api/v2/tasks")
@limiter.limit("10/minute")
async def create_task(request: Request, task: TaskRequest):
# ...欢迎贡献代码！请遵循以下流程：
点击右上角的 Fork 按钮
git checkout -b feature/amazing-featuregit add .
git commit -m "Add some amazing feature"git push origin feature/amazing-feature在 GitHub 上创建 Pull Request
- ✅ 代码通过所有测试
- ✅ 遵循代码规范
- ✅ 添加必要的文档
- ✅ 更新 CHANGELOG
- ✅ 没有引入新的安全问题
本项目采用 MIT 许可证。详见 LICENSE 文件。
感谢以下开源项目和服务：
- LangChain - 智能代理框架
- DeepSeek - 大语言模型
- ModelScope - MCP 服务器托管
- FastAPI - Web 框架
- React - 前端框架
- 问题反馈: GitHub Issues
- 功能建议: GitHub Discussions
- 邮件: your-email@example.com
- ✨ 采用 SimpleMCPEngine，简化架构
- ✨ 集成 25+ MCP 工具
- ✨ 添加实时可视化界面
- 🐛 修复工具调用失败问题
- 📝 完善文档
- 🎉 初始版本发布
- ✨ 基础的任务执行功能
- ✨ 支持 10+ MCP 工具
最后更新: 2025-12-18
版本: 2.0.0
维护者: AgenticAI Team
⭐ 如果这个项目对你有帮助，请给一个 Star！
Made with ❤️ by AgenticAI Team