大模型应用开发实战：基于LangChain构建企业级ReAct数据分析Agent

---

1. 为什么你的大模型应用需要一个Agent？

课程第一章节强调一个核心观点：纯粹的对话式LLM只是一个“大脑”，而Agent是赋予它“手脚”的关键。企业中的数据分析需求往往涉及多步推理——例如：“对比上季度华东区和华南区的销售额差异，并生成可视化报告”——这需要模型自主决定：

查询数据库获取原始数据；使用Python进行数据清洗和聚合；调用绘图库生成图表；最后用自然语言总结结论。

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

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

---

2. 系统总体架构与技术选型

2.1 业务场景定义

数据源：MySQL数据库中存储销售订单表（orders）、产品表（products）、区域表（regions），总计约500万行。用户需求：用自然语言提问，如“哪个产品线毛利率最高？”、“过去30天每日订单趋势”等。输出形式：文本回答 可交互的HTML图表（通过Base64嵌入）。

2.2 架构组件

代码语言：ja vascript

复制

------------------- --------------------- ---------------------- | 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监控。

---

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

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

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

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

代码语言：ja vascript

复制

from langchain.tools import StructuredToolfrom pydantic import BaseModel, Fieldimport pymysqlimport jsonclass 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。

代码语言：ja vascript

复制

import subprocess, tempfile, base64, osclass PythonInput(BaseModel):code: str = Field(description="Python3代码，必须将最终结果赋值给变量`result`，如果是图表则保存为`chart.png`")def execute_python(code: str) -> str:# 使用临时文件执行，限制资源with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:# 注入安全限制：禁用危险模块safe_code = """import matplotlibmatplotlib.use('Agg')import matplotlib.pyplot as pltimport pandas as pdimport numpy as np# 用户代码开始""" code """# 用户代码结束# 如果图表存在则编码返回if os.path.exists('chart.png'):with open('chart.png', 'rb') as img:b64 = base64.b64encode(img.read()).decode()result = f"data:image/png;base64,{b64}""""f.write(safe_code)try:proc = subprocess.run(["python3", f.name], timeout=30,capture_output=True, text=True, env={"PYTHONPATH": ""})os.unlink(f.name)if proc.returncode != 0:return f"执行错误: {proc.stderr}"# 从输出中提取result变量（实际可用exec捕获，此处简化）return proc.stdout[-1000:]# 实际应解析except Exception as e:return str(e)python_tool = StructuredTool.from_function(func=execute_python,name="python_analysis",description="执行Python代码用于数据分析和生成图表，代码中可使用pandas/numpy/matplotlib。")

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

代码语言：ja vascript

复制

from langchain_community.tools import DuckDuckGoSearchRunsearch_tool = DuckDuckGoSearchRun()# 包装为StructuredTool

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

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

代码语言：ja vascript

复制

from langchain.agents import create_react_agent, AgentExecutorfrom langchain.prompts import PromptTemplatefrom langchain_openai import ChatOpenAI# 使用腾讯云混元（兼容OpenAI接口）llm = ChatOpenAI(model="hunyuan-lite",openai_api_key=os.getenv("HUNYUAN_API_KEY"),openai_api_base="https://api.hunyuan.cloud.tencent.com/v1",temperature=0.1,streaming=True# 开启流式)# 自定义ReAct模板（增强指令）template = """You are a senior data analyst assistant. You ha ve access to these tools:{tools}Use the following format:Question: {input}Thought: you should always think about what to do next.Action: the action to take, must be one of [{tool_names}]Action Input: the input to the action in JSON formatObservation: the result of the action... (this Thought/Action/Action Input/Observation can repeat N times)Thought: I now know the final answerFinal Answer: the final answer to the original question, include chart if generated.Begin!Previous conversation history:{chat_history}Question: {input}{agent_scratchpad}"""prompt = PromptTemplate.from_template(template)agent = create_react_agent(llm, tools=[sql_tool, python_tool, search_tool], prompt=prompt)agent_executor = AgentExecutor(agent=agent,tools=tools,verbose=True,max_iterations=8, # 防止死循环early_stopping_method="generate",return_intermediate_steps=True,handle_parsing_errors=True)

关键优化点：

流式输出：AgentExecutor 不支持原生流式，这里通过 astream_events 方法（LangChain 0.3 ）实现逐Token推送，大幅提升用户体验。对话记忆：使用 ConversationBufferWindowMemory 保留最近5轮对话，避免上下文超长。代码语言：ja vascript

复制

from langchain.memory import ConversationBufferWindowMemorymemory = ConversationBufferWindowMemory(k=5, memory_key="chat_history", return_messages=True)agent_executor.memory = memory

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

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

代码语言：ja vascript

复制

from fastapi import FastAPI, BackgroundTasksfrom sse_starlette.sse import EventSourceResponseimport asyncioapp = FastAPI()@app.post("/agent/stream")async def agent_stream(query: str, session_id: str):# 从Redis加载该session的历史memory.load_memory_variables({"session_id": session_id})async def event_generator():async for event in agent_executor.astream_events({"input": query},version="v1"):if event["event"] == "on_chain_stream":# 流式输出tokenyield {"data": event["data"]["chunk"], "event": "token"}elif event["event"] == "on_tool_start":yield {"data": f"调用工具: {event['name']}", "event": "tool"}elif event["event"] == "on_chain_end":yield {"data": "[DONE]", "event": "done"}# 保存新的对话到Redismemory.sa ve_context({"input": query}, {"output": final_answer})return EventSourceResponse(event_generator())

---

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

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，避免重复查询数据库。代码语言：ja vascript

复制

from hashlib import md5cache_key = md5(sql_query.encode()).hexdigest()cached = redis_client.get(cache_key)if cached:return cached.decode()# 否则执行并缓存

4.3 监控与告警

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

每次请求的Token消耗工具调用次数分布平均响应时间（P50/P95）

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

---

5. 实验效果与实战体会

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

指标	数值
任务完成率（最终给出正确结论）	86.5%
平均迭代步数	3.2步
平均总耗时（含LLM推理）	6.8s
工具调用准确率（选择正确工具）	94%

典型错误分析：

模型偶尔生成不存在的列名（如product_line写成product_line_name），通过在SQL工具中增加错误捕获并返回列不存在信息，模型可在下一步自行修正——这验证了ReAct的自纠错能力。长对话时，记忆窗口截断导致丢失关键上下文，后续版本计划引入 ConversationSummaryMemory 替代。

成本测算：每次问答平均消耗约1500个input token和300个output token（混元模型），按腾讯云刊例价约为0.02元/次，远低于雇佣人工数据分析师的成本。

---

6. 总结与展望

本文基于【实战课程】的核心理念，完整呈现了一个生产级数据分析Agent的开发全链路。通过精心设计的工具、合理的提示工程、流式交互，大模型完全能够承担企业级数据分析任务。同时，腾讯云提供的容器、日志、监控服务为系统的稳定运行提供了坚实底座。