可控大模型应用框架 simple_ai_toolset

# 定义 qa prompt
qa_prompt_tmpl_str = (
    "## 角色\n"
    "请你扮演中国软件水平考试高级辅导专家，负责用户发送的概念讲解和发送的题目解答。\n"

    "## 技能\n"
    "### 技能 1：概念讲解\n"
    "当我发送一些软考概念题目的上下文信息。\n"
    "Step1：根据概括帮我讲解一下相关内容，讲解时尽量通俗易懂，并给出恰当的例子，优先使用 markdown 表格的形式来呈现\n"
    "Step2：出 2 道相关的选择题，在出完题目的最后给出答案和对答案的详细讲解。\n"

    "输出格式为：\n"
    "=====\n"

    "# 一、AI 讲解\n"
    "<概念讲解>\n"

    "# 二、AI 出题\n"
    "### （1）题目\n"
    "<出对应的 2 道选择题>\n"
    "### （2）答案和解析\n"
    "<所有选择题的答案和解释，每个答案和对应解释放在一起>\n"

    "=====\n"

    "## 要求\n"
    "1 必须使用中文回答我\n"
    "2 解答时，尽量使用通俗易懂的语言\n"
    "3 讲解时，如果有可能尽量给出相关例子\n"
    "4 讲解时，优先考虑使用 markdown 表格的方式呈现，如果出现不同层级的概念，可以将不同层级的概念用不同的表格表示\n"
    "5 给出答案和解析时，每道题的答案和解释要在一起给出，答案的解释需要详尽\n"
    "上下文信息如下。\n"
    "---------------------\n"
    "{context_str}\n"
    "---------------------\n"
    "请根据上下文信息而不是先验知识来回答以下的查询。回复格式：markdown 。 \n"
    "作为一个中国软考考试专家人工智能助手，你的回答要尽可能严谨。\n"
    "Query: 关于 {query_str}\n"
    "Answer: "
)
qa_prompt_tmpl = PromptTemplate(qa_prompt_tmpl_str)
# 更新查询引擎中的 prompt template
query_engine.update_prompts(
    {"response_synthesizer:text_qa_template": qa_prompt_tmpl,
     "response_synthesizer:refine_template": refine_prompt_tmpl}
)

response = query_engine.query(question)

3. Agent 智能体代理

Agent（智能体）将成为未来程序员开发的核心工作流。由于涉及大量模糊逻辑和定制化需求，Agent 的开发价值巨大。本框架提供了最实用的智能体实现，强调简化与可控性，使其能无缝整合到项目开发中。

框架利用 ReAct（Reasoning + Acting）思想，让智能体自主选择工具、反思行动结果，并逐步完成任务。

ReAct Prompt 示例：

prompt_template = """
                You run in a loop of Thought, Action, PAUSE, Action_Response.
                At the end of the loop you output an Answer.

                Use Thought to understand the question you have been asked.
                Use Action to run one of the actions available to you - then return PAUSE.
                Action_Response will be the result of running those actions.

                Your available actions are:

                {actions_list}

                To use an action, please use the following format:

                Thought: Do I need to use a tool? Yes

                Action:

                {{
                    "function_name": tool_name,
                    "function_params": {{
                        "param": "value"
                    }}
                }}

                Action_Response: the result of the action"""

智能体工具调用示例：

@router.get("/ai_agent_lego")
async def ai_agent_lego(question: str, request: Request):
    
    llm_instance = LLM.create(provider=LLMProvider.OPENAI, model_name="gpt-4o")
    # Create an agent instance
    agent = Agent(LLMProvider.OPENAI, model_name="gpt-4o")

    #add tools
    agent.add_tool(tool_brick4_search)
    agent.add_tool(tool_smzdm)
    agent.add_tool(tool_brick4)

    user_query = question

    # Generate a response
    final_result = agent.generate_response(user_query)
    print(final_result)

框架应用场景

1. 可控的 RAG 应用

利用 LangChain 等框架往往存在'黑盒'问题，难以精确控制输入输出。本框架提供方法使事情变得可控，适合程序员将 AI 整合进业务流程。

documents = SimpleDirectoryReader(input_files=['./data/abc.txt']).load_data()
print(documents)

index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)

query_engine = index.as_query_engine()
# 可以指定固定回复格式
# query_engine = index.as_query_engine(output_cls=BlogTitles) 

# 定义 qa prompt
qa_prompt_tmpl_str = (
    "上下文信息如下。\n"
    "---------------------\n"
    "{context_str}\n"
    "---------------------\n"
    "请根据上下文信息而不是先验知识来回答以下的查询。回复格式：请务必把上下文的图片 image jpg 放在回答的底部作为参考。"
    "作为一个油画艺术人工智能助手，你的回答要尽可能严谨。\n"
    "Query: {query_str}\n"
    "Answer: "
)
qa_prompt_tmpl = PromptTemplate(qa_prompt_tmpl_str)

# 定义 refine prompt
refine_prompt_tmpl_str = (
    "原始查询如下：{query_str}"
    "我们提供了现有答案：{existing_answer}"
    "我们有机会通过下面的更多上下文来完善现有答案（仅在需要时）。"
    "------------"
    "{context_msg}"
    "------------"
    "考虑到新的上下文，优化原始答案以更好地回答查询。 如果上下文没有用，请返回原始答案。"
    "Refined Answer:"
)
refine_prompt_tmpl = PromptTemplate(refine_prompt_tmpl_str)

# 更新查询引擎中的 prompt template
query_engine.update_prompts(
    {"response_synthesizer:text_qa_template": qa_prompt_tmpl,
     "response_synthesizer:refine_template": refine_prompt_tmpl}
)

response = query_engine.query(question)

2. 智能体问答助手

实现让智能体作为大模型的问答助手，自行判断用户意图，调用不同接口 API 和服务获取知识。例如积木问答机器人，用户可以询问积木新闻、搜索、新品、优惠活动等，智能体自行判断调用相应函数。

本框架与 LangChain 的不同在于，Agent 完全透明可控，代码和提示词均可自行修改，方便程序员实现可控编程。

#Define a brick4 search tool  定义工具，依靠注释让 agent 自动判断调用哪个 tool
def tool_brick4_search(keyword: str):
    """
    Search content from a given Keyword.
    Parameters: keyword (str)
    """
    llm_instance = LLM.create(provider=LLMProvider.OPENAI, model_name="gpt-4o") 
    search_query_encoded = quote_plus(keyword)
    sjina = str("http://brick4.com/get/set?filter_brandorder=0&filter_order=1&brandorder=1&page=1&s="+search_query_encoded)

@router.get("/ai_agent_lego")
async def ai_agent_lego(question: str, request: Request):
    agent = Agent(LLMProvider.OPENAI, model_name="gpt-4o")
    agent.add_tool(tool_brick4_search)
    agent.add_tool(tool_smzdm)
    agent.add_tool(tool_brick4)
    user_query = question
    final_result = agent.generate_response(user_query)

3. 写作机器人

当所有 Tools 都变为可控后，可以轻松实现类似 Coze 的控制流。Coze 等平台常因小意外报错停止，而本框架允许自行实现和控制流程判断，处理异常情况，提高可用性。

@router.get("/ai_write_lego")
async def ai_write_lego(request: Request):
    
    llm_instance = LLM.create(provider=LLMProvider.OPENAI, model_name="moonshot-v1-128k")
    
    tool_brick_news_str = tool_brick_news()
    main_points_1 = llm_instance.generate_response(prompt=f"从以下 [积木情报] 中提取关键信息，格式为：（标题，日期时间，链接，图片），上下文是：{tool_brick_news_str}")
    tool_brick_news2_str = tool_brick_news2()
    main_points_1b = llm_instance.generate_response(prompt=f"从以下 [积木情报] 中提取关键信息，格式为：（标题，日期时间，链接，图片），上下文是：{tool_brick_news2_str}")
   
    tool_smzdm_str = tool_smzdm()
    main_points_2 = llm_instance.generate_response(prompt=f"从以下 [积木优惠] 中提取关键信息，格式为：（标题，价格，折扣，评价，平台，图片），上下文是：{tool_smzdm_str}")
    tool_brick4_str = tool_brick4()
    main_points_3 = llm_instance.generate_response(prompt=f"从以下 [积木新品] 中提取关键信息，格式为：（名称，价格，上市日期，平台，图片）: {tool_brick4_str}")
    all_str ="\n\n # 今天积木情报：\n\n" + main_points_1 + "\n\n  " + main_points_1b + "\n\n # 今天积木优惠：\n\n" + main_points_2 + "\n\n # 今天积木新品：\n\n" + main_points_3
    
    main_points_best = llm_instance.generate_response(prompt=f"你是一个积木专栏作家，请给上下文内容一个总结，用中文回答，上下文如下：{all_str}")
    print(main_points_best)
    print(all_str)

架构设计与扩展

1. 目录结构规范

为了便于维护和扩展，建议采用以下目录结构：

simple_ai_toolset/
├── app/
│   ├── api/
│   │   └── v1/
│   │       ├── rag.py
│   │       └── agent.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── models/
│   │   └── db_models.py
│   └── services/
│       ├── embedding_service.py
│       └── llm_service.py
├── tests/
├── requirements.txt
└── Dockerfile

2. 性能优化与缓存

在大模型应用中，延迟和成本是关键指标。建议在框架中引入 Redis 缓存机制：

• Embedding 缓存：对于相同的文本片段，复用已生成的向量，避免重复调用 Embedding API。
• Prompt 缓存：对于固定的 System Prompt，预加载到内存中。
• Rate Limiting：除了 slowapi 的全局限速，可在 Service 层针对特定用户或 IP 实施更细粒度的限流。

3. 部署与运维

建议使用 Docker 容器化部署，配合 Kubernetes 进行编排。Dockerfile 示例：

FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

在生产环境中，应启用日志聚合（如 ELK Stack）和链路追踪（如 Jaeger），以便监控 AI 调用的耗时和错误率。

总结

本框架旨在解决 AI 大模型应用开发中的可控性问题，帮助程序员从处理信息转向处理知识。核心价值点如下：

1. 可控性：通过 Pydantic 和 Prompt 工程，确保输入输出的确定性。
2. 集成性：整合了 RAG、Agent、数据库等常用组件，节省开发时间。
3. 扩展性：模块化设计，支持快速添加新的 AI 功能和工具。
4. 实用性：面向真实业务场景，而非仅停留在 Demo 阶段。

通过本框架，开发者可以更高效地构建企业级 AI 应用，将大模型技术真正落地到业务流程中。