本示例展示了如何使用LlamaIndex的Workflows功能实现JSONalyze Query Engine该引擎能够对JSON数据进行SQL查询和分析。1. 案例目标JSONalyze Query Engine旨在处理API调用后返回的JSON数据并执行统计分析。具体目标包括将JSON数据加载到内存中的SQLite数据库使用大语言模型将自然语言问题转换为SQL查询执行SQL查询并返回结果基于查询结果生成自然语言回答2. 技术栈与核心依赖LlamaIndex- 用于构建工作流和查询引擎OpenAI API- 提供大语言模型服务SQLite- 内存数据库用于存储和查询JSON数据sqlite-utils- 用于将JSON数据加载到SQLite数据库主要依赖安装pip install -U llama-index pip install sqlite-utils3. 环境配置在开始之前需要设置OpenAI API密钥import os os.environ[OPENAI_API_KEY] sk-...4. 案例实现4.1 定义事件类首先定义一个自定义事件类JsonAnalyzerEvent用于在工作流步骤之间传递数据from llama_index.core.workflow import Event from typing import Dict, List, Any class JsonAnalyzerEvent(Event): Event containing results of JSON analysis. Attributes: sql_query (str): The generated SQL query. table_schema (Dict[str, Any]): Schema of the analyzed table. results (List[Dict[str, Any]]): Query execution results. sql_query: str table_schema: Dict[str, Any] results: List[Dict[str, Any]]4.2 定义提示模板定义用于生成SQL查询和合成回答的提示模板from llama_index.core.prompts.prompt_type import PromptType from llama_index.core.prompts import PromptTemplate DEFAULT_RESPONSE_SYNTHESIS_PROMPT_TMPL ( Given a query, synthesize a response based on SQL query results to satisfy the query. Only include details that are relevant to the query. If you dont know the answer, then say that.\n SQL Query: {sql_query}\n Table Schema: {table_schema}\n SQL Response: {sql_response}\n Query: {query_str}\n Response: ) DEFAULT_RESPONSE_SYNTHESIS_PROMPT PromptTemplate( DEFAULT_RESPONSE_SYNTHESIS_PROMPT_TMPL, prompt_typePromptType.SQL_RESPONSE_SYNTHESIS, ) DEFAULT_TABLE_NAME items4.3 实现工作流创建JSONAnalyzeQueryEngineWorkflow类包含两个主要步骤class JSONAnalyzeQueryEngineWorkflow(Workflow): step async def jsonalyzer( self, ctx: Context, ev: StartEvent ) - JsonAnalyzerEvent: 分析JSON数据并执行SQL查询 # 导入sqlite-utils包 try: import sqlite_utils except ImportError as exc: IMPORT_ERROR_MSG ( sqlite-utils is needed to use this Query Engine:\n pip install sqlite-utils ) raise ImportError(IMPORT_ERROR_MSG) from exc # 获取输入参数 await ctx.store.set(query, ev.get(query)) await ctx.store.set(llm, ev.get(llm)) query ev.get(query) table_name ev.get(table_name) list_of_dict ev.get(list_of_dict) prompt DEFAULT_JSONALYZE_PROMPT # 创建内存SQLite数据库并加载JSON数据 db sqlite_utils.Database(memoryTrue) try: db[ev.table_name].insert_all(list_of_dict) except sqlite_utils.utils.sqlite3.IntegrityError as exc: print_text( fError inserting into table {table_name}, expected format: ) print_text([{col1: val1, col2: val2, ...}, ...]) raise ValueError(Invalid list_of_dict) from exc # 获取表结构 table_schema db[table_name].columns_dict # 使用LLM生成SQL查询 response_str await ev.llm.apredict( promptprompt, table_nametable_name, table_schematable_schema, questionquery, ) sql_parser DefaultSQLParser() sql_query sql_parser.parse_response_to_sql(response_str, ev.query) # 执行SQL查询 try: results list(db.query(sql_query)) except sqlite_utils.utils.sqlite3.OperationalError as exc: print_text(fError executing query: {sql_query}) raise ValueError(Invalid query) from exc return JsonAnalyzerEvent( sql_querysql_query, table_schematable_schema, resultsresults ) step async def synthesize( self, ctx: Context, ev: JsonAnalyzerEvent ) - StopEvent: 基于查询结果合成回答 llm await ctx.store.get(llm, defaultNone) query await ctx.store.get(query, defaultNone) response_str llm.predict( DEFAULT_RESPONSE_SYNTHESIS_PROMPT, sql_queryev.sql_query, table_schemaev.table_schema, sql_responseev.results, query_strquery, ) response_metadata { sql_query: ev.sql_query, table_schema: str(ev.table_schema), } response Response(responseresponse_str, metadataresponse_metadata) return StopEvent(resultresponse)4.4 准备测试数据创建一个包含个人信息的JSON列表作为测试数据json_list [ { name: John Doe, age: 25, major: Computer Science, email: john.doeexample.com, address: 123 Main St, city: New York, state: NY, country: USA, phone: 1 123-456-7890, occupation: Software Engineer, }, # ... 更多数据 ]4.5 执行查询初始化工作流并执行查询# 初始化LLM llm OpenAI(modelgpt-3.5-turbo) # 创建工作流实例 w JSONAnalyzeQueryEngineWorkflow() # 执行查询 query What is the maximum age among the individuals? result await w.run( queryquery, list_of_dictjson_list, llmllm, table_nameDEFAULT_TABLE_NAME ) # 显示结果 display( Markdown( Question: {}.format(query)), Markdown(Answer: {}.format(result)), )5. 案例效果通过JSONalyze Query Engine我们可以对JSON数据执行各种查询例如最大值查询What is the maximum age among the individuals?条件计数How many individuals have an occupation related to science or engineering?模式匹配How many individuals have a phone number starting with 1 234?百分比计算What is the percentage of individuals residing in California (CA)?特定值计数How many individuals have a major in Psychology?6. 案例实现思路JSONalyze Query Engine的实现基于以下思路数据转换将JSON数据转换为SQLite数据库表使数据可以通过SQL查询自然语言到SQL的转换使用LLM将自然语言问题转换为SQL查询语句工作流编排使用LlamaIndex的Workflows功能编排数据处理和查询执行步骤结果合成将SQL查询结果转换为自然语言回答7. 扩展建议支持更多数据源扩展支持CSV、Excel等格式的数据源复杂查询支持增强对复杂SQL查询如JOIN、子查询的支持可视化支持添加数据可视化功能将查询结果以图表形式展示查询缓存实现查询结果缓存提高重复查询的响应速度多表关联支持多表关联查询处理更复杂的数据关系查询优化添加SQL查询优化功能提高查询效率8. 总结JSONalyze Query Engine示例展示了如何使用LlamaIndex的Workflows功能构建一个强大的数据分析工具。通过将JSON数据转换为SQLite数据库并利用LLM进行自然语言到SQL的转换用户可以使用自然语言对数据进行复杂的查询和分析。这种方法不仅降低了数据分析的技术门槛还提供了灵活的数据探索能力特别适合需要快速分析API返回数据的场景。