AI Agent 融合实战:智能数据分析助手——LangChain、Function Calling等的综合应用

mkdir 5-overallcd 5-overallpython -m venv venvsource venv/bin/activatepip install langchain langchain-openai langgraph pandas matplotlib python-dotenv

import pandas as pddata = {"month": ["Jan", "Jan", "Jan", "Feb", "Feb", "Feb", "Mar", "Mar", "Mar"],"region": ["East", "West", "South", "East", "West", "South", "East", "West", "South"],"product": ["A", "B", "A", "B", "A", "B", "A", "B", "A"],"sales": [1000, 1500, 1200, 1300, 1400, 1100, 2000, 1800, 1600],"profit": [200, 300, 240, 260, 280, 220, 400, 360, 320],"quantity": [100, 150, 120, 130, 140, 110, 200, 180, 160]}df = pd.DataFrame(data)df.to_csv("sales.csv", index=False)print(df)

from langchain_core.tools import tool@tooldef query_sales_data(question: str) -> str:"""    对销售数据执行分析查询。支持：    - 最高/最低值：如"销售额最高的月份"    - 总和/平均：如"总销售额"    - 分组统计：如"各地区销售额"    - 条件筛选：如"产品A在3月的销量""""global df    q = question.lower()if"最高"in q or"最大"in q:if"销售额"in q:            max_row = df.loc[df['sales'].idxmax()]returnf"销售额最高的记录是{max_row['month']}月，{max_row['region']}地区，销售额为{max_row['sales']}"elif"利润"in q:            max_row = df.loc[df['profit'].idxmax()]returnf"利润最高的记录是{max_row['month']}月，{max_row['region']}地区，利润为{max_row['profit']}"if"总销售额"in q:returnf"总销售额为{df['sales'].sum()}"if"总利润"in q:returnf"总利润为{df['profit'].sum()}"if"各地区"in q and"销售额"in q:        result = df.groupby('region')['sales'].sum().to_dict()returnf"各地区销售额：{result}"if"产品"in q and"销量"in q:for product in df['product'].unique():if product in q:returnf"{product}产品总销量为{df[df['product']==product]['quantity'].sum()}"returnf"无法理解问题：{question}"

import matplotlib.pyplot as plt@tooldef plot_sales_data(chart_type: str = "line", metric: str = "sales") -> str:"""    生成销售数据可视化图表。    参数:        chart_type: 'line'折线图 或 'bar'柱状图        metric: 'sales'销售额 或 'profit'利润    返回:        图表保存路径"""global df    plt.figure(figsize=(10, 6))if chart_type == "line":        monthly = df.groupby('month')[metric].sum()        monthly.plot(kind='line', marker='o')        plt.title(f"Monthly {metric.capitalize()} Trend")        plt.xlabel("Month")        plt.ylabel(metric.capitalize())elif chart_type == "bar":        regional = df.groupby('region')[metric].sum()        regional.plot(kind='bar')        plt.title(f"{metric.capitalize()} by Region")    img_path = f"sales_{metric}_{chart_type}.png"    plt.savefig(img_path)    plt.close()returnf"图表已保存为{img_path}"

@tooldef analyze_sales_trend(question: str) -> str:"""    对销售数据进行深度分析，生成业务洞察。    参数:        question: 分析问题，如"为什么利润下降了？""""global df    q = question.lower()if"利润下降"in q or"利润为什么"in q:        monthly_profit = df.groupby('month')['profit'].sum()iflen(monthly_profit) >= 2:            months = list(monthly_profit.index)            changes = []for i inrange(1, len(months)):                change = monthly_profit.iloc[i] - monthly_profit.iloc[i-1]                changes.append(f"{months[i]}月相比{months[i-1]}月：{change}")returnf"利润变化趋势：\n" + "\n".join(changes)return"请提供更具体的分析问题。"

from langchain_openai import ChatOpenAIfrom langchain.agents import create_agentfrom langgraph.graph import StateGraph, END, add_messagesfrom langgraph.checkpoint.memory import InMemorySaverfrom typing import TypedDict, Annotatedmodel = ChatOpenAI(model="gpt-4", temperature=0)agent = create_agent(    model=model,    tools=[query_sales_data, plot_sales_data, analyze_sales_trend],    system_prompt="""你是一个专业的数据分析助手。你可以：1. 使用 query_sales_data 回答数据查询问题2. 使用 plot_sales_data 生成图表3. 使用 analyze_sales_trend 进行深度分析回答要简洁、专业。""")class AnalyzerState(TypedDict):    messages: Annotated[list, add_messages]    context: dictdef run_agent(state: AnalyzerState):    result = agent.invoke({"messages": state["messages"]})return {"messages": [result["messages"][-1]]}graph = StateGraph(AnalyzerState)graph.add_node("agent", run_agent)graph.set_entry_point("agent")graph.add_edge("agent", END)checkpointer = InMemorySaver()graph = graph.compile(checkpointer=checkpointer)

def chat():    thread_id = "data_analyzer_001"    config = {"configurable": {"thread_id": thread_id}}whileTrue:        user_input = input("\n👤 你: ")if user_input.lower() == 'exit':break        result = graph.invoke(            {"messages": [{"role": "user", "content": user_input}]},            config=config        )print(f"🤖 助手: {result['messages'][-1].content}")if __name__ == "__main__":    chat()

👤 你: 总销售额是多少？🤖 助手: 总销售额为12900。👤 你: 画一张销售额的月度趋势图🤖 助手: 图表已保存为sales_sales_line.png👤 你: 为什么利润下降了？🤖 助手: ...利润变化趋势：Feb月相比Jan月：20，Mar月相比Feb月：320👤 你: 那利润最高的地区是哪里？🤖 助手: 利润最高的地区是East

{"tool": "query_sales_data","arguments": {"question": "总销售额是多少？"}}

class PlanState(TypedDict):    input: str    plan: list    past_steps: list    final_answer: strdef planner(state: PlanState):"""规划节点：生成执行计划"""    prompt = f"为以下任务生成步骤列表：{state['input']}"    response = model.invoke(prompt)    lines = response.content.strip().split('\n')    plan = [line.strip() for line in lines if line.strip() andnot line.startswith('步骤')]return {"plan": plan}def executor(state: PlanState):"""执行节点：按顺序执行步骤"""ifnot state["plan"]:return {}    step = state["plan"][0]    result = agent.invoke({"messages": [{"role": "user", "content": f"执行：{step}"}]})    result_content = result["messages"][-1].content    new_past_steps = state.get("past_steps", []) + [(step, result_content)]    remaining_plan = state["plan"][1:]    final_answer = ""ifnot remaining_plan:        final_answer = "\n".join([f"{s}: {r}"for s, r in new_past_steps])return {"plan": remaining_plan, "past_steps": new_past_steps, "final_answer": final_answer}

def should_continue(state: PlanState) -> str:return"continue"if state.get("plan") else"end"plan_graph = StateGraph(PlanState)plan_graph.add_node("planner", planner)plan_graph.add_node("executor", executor)plan_graph.set_entry_point("planner")plan_graph.add_edge("planner", "executor")plan_graph.add_conditional_edges("executor", should_continue,    {"continue": "executor", "end": END})plan_app = plan_graph.compile()

mkdir -p ~/.openclaw/skills/data_analyzercd ~/.openclaw/skills/data_analyzertouch SKILL.md __init__.py

---name: data_analyzerdescription: 销售数据分析技能，支持查询、图表生成和深度分析---# data_analyzer 技能## query_sales_data对销售数据执行分析查询。支持：最高销售额/利润、总销售额/总利润、各地区/各月份统计、产品销量筛选。参数: question (string) 用户的分析问题## plot_sales_data生成销售数据可视化图表。参数: chart_type (string) 'line'折线图 或 'bar'柱状图；metric (string) 'sales'销售额 或 'profit'利润## analyze_sales_trend对销售数据进行深度分析，生成业务洞察。参数: question (string) 分析问题

from openclaw.skill import Skillimport pandas as pdimport matplotlib.pyplot as pltimport osdf = pd.read_csv(os.path.join(os.path.dirname(__file__), "sales.csv"))class DataAnalyzerSkill(Skill):def __init__(self):        super().__init__()self.df = dfdef query_sales_data(self, question: str) -> str:        q = question.lower()if"最高"in q or"最大"in q:if"销售额"in q:                max_row = self.df.loc[self.df['sales'].idxmax()]returnf"销售额最高的记录是{max_row['month']}月，{max_row['region']}地区，销售额为{max_row['sales']}"if"总销售额"in q:returnf"总销售额为{self.df['sales'].sum()}"returnf"无法理解：{question}"def plot_sales_data(self, chart_type: str = "line", metric: str = "sales") -> str:        plt.figure(figsize=(10, 6))if chart_type == "line":self.df.groupby('month')[metric].sum().plot(kind='line', marker='o')elif chart_type == "bar":self.df.groupby('region')[metric].sum().plot(kind='bar')        img_path = f"/tmp/sales_{metric}_{chart_type}.png"        plt.savefig(img_path)        plt.close()returnf"图表已保存为{img_path}"def analyze_sales_trend(self, question: str) -> str:        q = question.lower()if"利润下降"in q or"利润为什么"in q:            monthly_profit = self.df.groupby('month')['profit'].sum()iflen(monthly_profit) >= 2:                months = list(monthly_profit.index)                changes = [f"{months[i]}月相比{months[i-1]}月：{monthly_profit.iloc[i]-monthly_profit.iloc[i-1]}"for i inrange(1, len(months))]returnf"利润变化趋势：\n" + "\n".join(changes)return"请提供更具体的分析问题。"skill = DataAnalyzerSkill()

cp /path/to/sales.csv ~/.openclaw/skills/data_analyzer/openclaw config set skills.load.extraDirs '["~/.openclaw/skills/data_analyzer"]'openclaw gateway restartopenclaw skills list

👤 你: 使用 data_analyzer 技能查询：总销售额是多少？🤖 助手: 总销售额为12900。👤 你: 画一张销售额的月度趋势图🤖 助手: 图表已保存为...👤 你: 那利润最高的地区是哪里？🤖 助手: 利润最高的地区是East

data_analyzer/├── SKILL.md        # 自然语言描述，给模型看├── __init__.py     # Python 实现，给机器跑└── sales.csv       # 数据文件，跟技能绑在一起