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

+-------------------+     +---------------------+     +----------------------+
|   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 |
+-------------------+

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
)

import subprocess, tempfile, base64, os

class 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 matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import pandas as pd
import 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。"
)

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

from langchain.agents import create_react_agent, AgentExecutor
from langchain.prompts import PromptTemplate
from 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 have 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 format
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final 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
)

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

from fastapi import FastAPI, BackgroundTasks
from sse_starlette.sse import EventSourceResponse
import asyncio

app = 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":
                # 流式输出token
                yield {"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"}
        # 保存新的对话到Redis
        memory.save_context({"input": query}, {"output": final_answer})
    
    return EventSourceResponse(event_generator())

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