基于LangChain构建企业级ReAct数据分析Agent

传统的RAG只能检索静态文本，无法动态执行计算。而Agent通过工具调用（Tool Calling） 与推理-行动循环（ReAct），将复杂任务拆解为可执行的子步骤，这正是大模型应用开发从“玩具”走向“生产力”的关键一跃。​

本文将以一个智能销售数据分析助手为例，完整演示如何基于LangChain + OpenAI兼容API（也可替换为开源Qwen模型）构建高可靠性的Agent系统，并部署在腾讯云TKE集群上，支撑日均万次查询。​

​

​

系统总体架构与技术选型​

2.1 业务场景定义​

•

数据源：MySQL数据库中存储销售订单表（orders）、产品表（products）、区域表（regions），总计约500万行。​

•

用户需求：用自然语言提问，如“哪个产品线毛利率最高？”、“过去30天每日订单趋势”等。​

•

输出形式：文本回答 + 可交互的HTML图表（通过Base64嵌入）。​

2.2 架构组件​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

+-------------------+ +---------------------+ +----------------------+| Web UI (React) |---->| API Gateway (Kong) |---->| Agent Orchestrator || (Streaming Chat) | | Auth + Rate Limit | | (LangChain) |+-------------------+ +---------------------+ +----------------------+|​

v​

+-------------------+ +---------------------+ +----------------------+| Tool Executors |<----| Memory (Redis) |<----| LLM Router || - SQL Executor | | (Conversation ctx) | | (OpenAI / 国产模型) || - Python Sandbox | +---------------------+ +----------------------+| - Web Search API |+-------------------+​

​

关键选型决策：​

•

Agent框架：LangChain 0.3+ 的 create_react_agent，配合 AgentExecutor 实现流式处理。​

•

LLM：采用腾讯云混元大模型（或OpenAI GPT-4o-mini），通过统一 ChatOpenAI 接口接入，便于切换。​

•

工具隔离：Python执行器使用 RestrictedPython 或 docker-exec 沙箱，保证安全；SQL执行器仅赋予只读权限。​

•

缓存层：Redis存储对话历史（窗口大小为20条），并缓存常见查询的中间结果（TTL=600s）。​

•

部署：腾讯云容器服务TKE，配合CLS日志采集和Prometheus监控。​

​

​

核心实现：从工具定义到Agent编排​

3.1 定义工具（Tools）—— 让模型学会“动手”​

每个工具必须包含名称、描述、参数schema，模型会根据用户问题决定调用哪个工具、传入什么参数。我们定义三个核心工具：​

① SQL查询工具：接收自然语言生成的SQL（由模型自己生成），执行并返回结果。​

​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

from langchain.tools import StructuredTool​

from pydantic import BaseModel, Field​

import pymysql​

import json​

​

class SQLInput(BaseModel):sql_query: str = Field(description="有效的MySQL SELECT语句")limit: int = Field(default=100, description="返回行数上限")​

​

def execute_sql(sql_query: str, limit: int = 100) -> str:​

# 使用连接池，只读事务​

conn = pymysql.connect(​

host=os.getenv("DB_HOST"), user=os.getenv("DB_USER"),​

password=os.getenv("DB_PASS"), database="sales_db",​

charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor​

)with conn.cursor() as cur:​

# 强制增加LIMIT防止拖库​

if "limit" not in sql_query.lower():​

sql_query += f" LIMIT {limit}"​

cur.execute(sql_query)​

rows = cur.fetchall()​

conn.close()​

# 截断过长的返回结果（防止token爆炸）​

return json.dumps(rows[:limit], ensure_ascii=False)[:3000]​

​

sql_tool = StructuredTool.from_function(​

func=execute_sql,​

name="sql_query",​

description="执行只读SQL查询并返回JSON结果，用于获取销售数据、产品信息等。",​

args_schema=SQLInput​

)​

​

② Python数据分析工具：允许模型编写Python代码进行聚合、统计、绘图，并返回图表Base64。​

​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

import subprocess, tempfile, base64, os​

​

​

③ 外部知识搜索工具：当内部数据不足时，可搜索公开信息（如行业平均毛利率），使用腾讯云ES或Bing Search API。​

​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

from langchain_community.tools import DuckDuckGoSearchRun​

​

3.2 构建ReAct Agent —— 核心编排逻辑​

LangChain的 create_react_agent 使用经典的 Thought/Action/Observation 循环。我们自定义了系统提示模板，引导模型分步推理：​

​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

from langchain.agents import create_react_agent, AgentExecutor​

from langchain.prompts import PromptTemplate​

from langchain_openai import ChatOpenAI​

​

# 使用腾讯云混元（兼容OpenAI接口）​

llm = ChatOpenAI(​

​

关键优化点：​

•

流式输出：AgentExecutor 不支持原生流式，我们通过 astream_events 方法（LangChain 0.3+）实现逐Token推送，大幅提升用户体验。​

•

对话记忆：使用 ConversationBufferWindowMemory 保留最近5轮对话，避免上下文超长。​

​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

from langchain.memory import ConversationBufferWindowMemory​

memory = ConversationBufferWindowMemory(k=5, memory_key="chat_history", return_messages=True)​

agent_executor.memory = memory​

​

3.3 流式响应实现（FastAPI后端）​

为了让前端实时看到Agent的思考过程，我们在FastAPI中实现SSE（Server-Sent Events）：​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

from fastapi import FastAPI, BackgroundTasks​

from sse_starlette.sse import EventSourceResponse​

import asyncio​

​

​

​

​

部署与性能调优（腾讯云实践）​

4.1 容器化与资源限制​

编写Dockerfile，基于 python:3.11-slim，安装依赖并设置非root用户。在TKE上部署时，每个Pod分配：​

•

CPU: 4核​

•

内存: 8Gi​

•

环境变量通过Secret注入（数据库密码、API Key）​

使用HPA（Horizontal Pod Autoscaler）根据CPU利用率（阈值70%）动态伸缩，最少2个副本，最多10个副本。​

4.2 缓存与成本控制​

大模型的调用费用是主要成本。我们引入两层缓存：​

•

语义缓存：使用 redisvl 库对用户问题做embedding，相似度>0.95时直接返回历史答案（无需调用LLM）。命中率约25%。​

•

工具结果缓存：SQL查询结果以 sql_hash 为键存入Redis，避免重复查询数据库。​

​

代码语言：javascript​

AI代码解释​

​

代码块​

JavaScript

​

4.3 监控与告警​

接入腾讯云CLS（日志服务）收集Agent的每一步 Thought 和 Observation，便于排查错误。同时使用Prometheus记录：​

•

每次请求的Token消耗​

•

工具调用次数分布​

•

平均响应时间（P50/P95）​

设置告警规则：当错误率超过5%或P95延迟 > 8s 时触发钉钉通知。​

​

​

实验效果与实战体会​

我们使用Hollis课程内测阶段的200个真实销售分析问题对系统进行压测，结果如下：​

​

​

典型错误分析：​

•

模型偶尔生成不存在的列名（如product_line写成product_line_name），通过在SQL工具中增加错误捕获并返回列不存在信息，模型可在下一步自行修正——这验证了ReAct的自纠错能力。​

•

长对话时，记忆窗口截断导致丢失关键上下文，后续版本计划引入 ConversationSummaryMemory 替代。​