LQAgentPlatform/config/模型调用指南.md

统一模型调用指南

概述

本项目采用统一的模型配置管理，所有模型配置集中存储在 model_setting.yaml 中，通过 model_config_loader.py 提供统一接口。

配置文件结构

model_setting.yaml

# 可用模型列表（必须与 model_handler.py 中的模型类型名称一致）
available_models:
  - qwen3_5_35b_a3b        # DashScope Qwen3.5-35B-A3B
  - shutian_qwen3_5_35b    # 蜀天Qwen3.5-35B
  - shutian_qwen3_5_122b   # 蜀天Qwen3.5-122B
  - lq_qwen3_8b_emd        # 本地Embedding模型

# 功能模块模型配置
model_settings:
  # 文档分类 - 二级分类
  doc_classification_secondary:
    model: shutian_qwen3_5_35b
    enable_thinking: false
    description: "文档二级分类，蜀天35B"

  # 文档分类 - 三级分类
  doc_classification_tertiary:
    model: shutian_qwen3_5_35b
    enable_thinking: false
    description: "文档三级分类，蜀天35B"

  # 完整性审查 - 内容生成
  completeness_review_generate:
    model: shutian_qwen3_5_122b
    enable_thinking: true
    description: "完整性审查内容生成，蜀天122B详细推理"

  # 敏感信息检查
  sensitive_check:
    model: shutian_qwen3_5_35b
    enable_thinking: false
    description: "敏感信息快速检查，蜀天35B"

  # ... 其他功能配置

# 默认配置（当功能未指定时使用）
default:
  model: shutian_qwen3_5_35b
  enable_thinking: false

模型调用方式

如何选择同步/异步版本

场景	推荐版本	说明
异步函数/协程中	await get_model_generate_invoke()	标准异步调用
同步函数/普通代码中	get_model_generate_invoke_sync()	同步阻塞调用
已有事件循环中	get_model_generate_invoke_sync()	避免嵌套事件循环问题

快速判断：

• 如果你的代码在 async def 函数中 → 使用异步版本
• 如果你的代码在普通 def 函数中 → 使用同步版本

方式一：使用 function_name（推荐 - 异步版本）

通过功能名称自动从 model_setting.yaml 加载对应的模型和 thinking 模式。

from foundation.ai.agent.generate.model_generate import generate_model_client

# 调用模型，自动加载 doc_classification_tertiary 配置的模型
response = await generate_model_client.get_model_generate_invoke(
    trace_id="my_trace_id",
    system_prompt="你是专家",
    user_prompt="请分析...",
    function_name="doc_classification_tertiary"  # 功能名称
)

适用场景：异步上下文（如 FastAPI 请求处理器、异步任务）。

方式一（同步）：使用 function_name（推荐 - 同步版本）

在同步上下文中使用，功能与异步版本完全一致。

from foundation.ai.agent.generate.model_generate import generate_model_client

# 同步调用（用于普通函数、同步上下文）
response = generate_model_client.get_model_generate_invoke_sync(
    trace_id="my_trace_id",
    system_prompt="你是专家",
    user_prompt="请分析...",
    function_name="doc_classification_tertiary"  # 功能名称
)

适用场景：同步上下文（如普通函数、已有事件循环环境中）。

注意：同步版本不支持 timeout 参数（使用构造时的默认值）。

方式二：使用 model_name（异步版本）

直接指定模型名称，跳过 model_setting.yaml 配置。

from foundation.ai.agent.generate.model_generate import generate_model_client

# 直接指定模型
response = await generate_model_client.get_model_generate_invoke(
    trace_id="my_trace_id",
    system_prompt="你是专家",
    user_prompt="请分析...",
    model_name="shutian_qwen3_5_122b"  # 直接指定模型
)

适用场景：需要临时切换模型，或测试特定模型性能。

方式二（同步）：使用 model_name（同步版本）

在同步上下文中直接指定模型名称。

from foundation.ai.agent.generate.model_generate import generate_model_client

# 同步调用，直接指定模型
response = generate_model_client.get_model_generate_invoke_sync(
    trace_id="my_trace_id",
    system_prompt="你是专家",
    user_prompt="请分析...",
    model_name="shutian_qwen3_5_122b"  # 直接指定模型
)

适用场景：同步上下文中需要临时切换模型。

方式三：使用 model_handler

通过 model_handler 获取模型实例，进行更底层的操作。

from foundation.ai.models.model_handler import model_handler

# 根据功能名称获取模型
model = model_handler.get_model_by_function("doc_classification_tertiary")

# 或根据模型名称获取
model = model_handler.get_model_by_name("shutian_qwen3_5_35b")

# 调用模型
response = await model.ainvoke(messages)

适用场景：需要自定义调用逻辑，或集成到现有框架。

方式四：流式调用

使用流式输出生成文本。

from foundation.ai.agent.generate.model_generate import generate_model_client

# 流式调用（使用 function_name）
for chunk in generate_model_client.get_model_generate_stream(
    trace_id="my_trace_id",
    messages=messages,
    function_name="rag_answer_generate"
):
    yield chunk

适用场景：实时响应场景，如聊天、长文本生成。

配置加载接口

获取模型配置

from foundation.ai.models.model_config_loader import (
    get_model_for_function,
    get_thinking_mode_for_function,
    get_full_config_for_function
)

# 获取指定功能的模型名称
model_name = get_model_for_function("doc_classification_tertiary")
# 返回: "shutian_qwen3_5_35b"

# 获取指定功能的 thinking 模式
thinking = get_thinking_mode_for_function("doc_classification_tertiary")
# 返回: False

# 获取完整配置
config = get_full_config_for_function("doc_classification_tertiary")
# 返回: ModelFunctionConfig(model="shutian_qwen3_5_35b", enable_thinking=False, description="...")

获取默认配置

from foundation.ai.models.model_config_loader import get_model_for_function

# 获取默认模型（当功能未指定时使用）
default_model = get_model_for_function("default")
# 返回: "shutian_qwen3_5_35b"

功能名称列表

功能名称	说明	默认模型
doc_classification_secondary	文档二级分类	shutian_qwen3_5_35b
doc_classification_tertiary	文档三级分类	shutian_qwen3_5_35b
doc_classification_tertiary_complex	三级分类-复杂段落	shutian_qwen3_5_122b
completeness_review_generate	完整性审查-生成	shutian_qwen3_5_122b
completeness_review_classify	完整性审查-分类	shutian_qwen3_5_35b
rag_query_understand	RAG查询理解	shutian_qwen3_5_35b
rag_answer_generate	RAG答案生成	shutian_qwen3_5_122b
sensitive_check	敏感信息检查	shutian_qwen3_5_35b
grammar_check	语法检查	shutian_qwen3_5_35b
timeliness_review	时效性审查	shutian_qwen3_5_35b
reference_review	规范性审查	shutian_qwen3_5_35b
directory_extraction	目录提取	shutian_qwen3_5_35b
default	默认兜底配置	shutian_qwen3_5_35b

迁移指南

从旧代码迁移

旧代码（硬编码模型）：

# 不推荐：硬编码模型名称
response = await model_client.get_model_generate_invoke(
    trace_id="xxx",
    messages=messages,
    model_name="qwen3_30b"  # 硬编码
)

新代码（使用配置）：

# 推荐：使用 function_name 从配置加载
response = await model_client.get_model_generate_invoke(
    trace_id="xxx",
    messages=messages,
    function_name="completeness_review_classify"  # 从配置加载
)

新增功能配置步骤

1. 在 model_setting.yaml 中添加配置：

   model_settings:
   # 新功能配置
   my_new_feature:
   model: shutian_qwen3_5_35b
   enable_thinking: false
   description: "新功能描述"
   

2. 在代码中使用：

异步版本：

response = await generate_model_client.get_model_generate_invoke(
    trace_id="xxx",
    messages=messages,
    function_name="my_new_feature"
)

同步版本：

response = generate_model_client.get_model_generate_invoke_sync(
    trace_id="xxx",
    messages=messages,
    function_name="my_new_feature"
)

注意事项

1. 优先使用 function_name：便于统一管理和调整模型配置
2. 不要随意修改 available_models：必须与 model_handler.py 中的模型类型名称一致
3. 保留 default 配置：作为兜底方案，防止功能未指定时出错
4. thinking 模式：仅对 Qwen3.5 系列模型有效，其他模型自动忽略
5. 同步/异步选择：
   • 异步版本 (get_model_generate_invoke)：适用于 async def 函数
   • 同步版本 (get_model_generate_invoke_sync)：适用于普通 def 函数，避免嵌套事件循环
6. 在已有事件循环中使用：如果你在 Jupyter Notebook 或其他已有事件循环的环境中，使用同步版本

同步/异步选择决策树

你在什么上下文中？
├── async def 函数中
│   └── 使用 await generate_model_client.get_model_generate_invoke(...)
├── def 普通函数中
│   └── 使用 generate_model_client.get_model_generate_invoke_sync(...)
└── 不确定/已有事件循环（如 Jupyter）
    └── 使用 generate_model_client.get_model_generate_invoke_sync(...)

故障排查

模型加载失败

检查日志：

[模型调用] 加载功能配置失败 [xxx]: ...

解决方案：

1. 检查 model_setting.yaml 是否存在
2. 检查功能名称是否拼写正确
3. 检查配置的模型是否在 available_models 列表中

配置未生效

检查代码是否正确传入了 function_name，而不是硬编码 model_name。

同步/异步版本选择错误

错误现象：RuntimeWarning: coroutine was never awaited

原因：在同步函数中使用了异步版本的调用。

解决方案：

# 错误：在普通函数中使用异步版本
def my_sync_function():
    response = await generate_model_client.get_model_generate_invoke(...)  # ❌

# 正确：在普通函数中使用同步版本
def my_sync_function():
    response = generate_model_client.get_model_generate_invoke_sync(...)   # ✓

错误现象：RuntimeError: This event loop is already running

原因：在已有事件循环的环境中（如 Jupyter Notebook）尝试运行异步代码。

解决方案：使用同步版本 get_model_generate_invoke_sync。

嵌套事件循环问题

错误现象：asyncio.run() cannot be called from a running event loop

原因：在异步函数内部尝试使用 asyncio.run() 运行另一个异步调用。

解决方案：直接使用 await 调用异步版本，或将内部调用改为同步版本。

相关文件

• config/model_setting.yaml - 模型配置文件
• config/model_config_loader.py - 配置加载接口
• foundation/ai/models/model_handler.py - 模型管理器
• foundation/ai/agent/generate/model_generate.py - 模型调用客户端（包含同步/异步版本）

API 快速参考

方法	类型	使用场景
await get_model_generate_invoke(...)	异步	async def 函数中
get_model_generate_invoke_sync(...)	同步	普通 def 函数中
get_model_generate_stream(...)	同步生成器	流式输出场景