Ollama AsyncClient仅处理2个并行实例问题排查

你遇到的并行数未达到设置值的问题，可能由配置未生效、系统资源限制或代码细节遗漏导致，以下是具体分析和排查方案：

一、先确认环境变量是否真正生效 #

OLLAMA_NUM_PARALLEL需要在Ollama服务启动前设置才会生效，如果是启动服务后再设置的环境变量，服务不会自动加载。请按以下步骤验证：

1. 关闭当前运行的Ollama服务
2. 设置环境变量：export OLLAMA_NUM_PARALLEL=4（Linux/macOS）或set OLLAMA_NUM_PARALLEL=4（Windows命令行）
3. 重新启动Ollama服务
4. 执行ollama env命令，查看输出中是否包含OLLAMA_NUM_PARALLEL=4，确认配置已生效

二、检查系统资源是否成为瓶颈 #

即使配置了并行数，Ollama的实际并行能力受限于硬件资源：

• GPU显存：phi3这类模型（尤其是非量化版本）每个实例会占用大量显存，如果你的GPU显存不足以同时运行4个实例，Ollama会自动限制并行数，避免内存溢出。可以在运行脚本时，用nvidia-smi（NVIDIA GPU）或系统任务管理器监控显存占用情况。
• CPU/内存：如果使用CPU推理，高并行数会占用大量CPU和内存资源，当资源耗尽时，系统会强制串行处理任务。

三、排查代码中的潜在问题 #

你的测试脚本存在一处可能影响并行观察的细节：
在main函数中，初始任务的创建虽然用了asyncio.create_task，但你通过await add_query逐个添加任务——虽然add_query内部的任务会立即进入事件循环，但await会等待当前add_query函数执行完毕（这一步很快，但可能导致任务启动有微小延迟）。可以修改为批量创建任务，确保所有任务同时进入调度：

async def main():
    model = "phi3"
    
    queries = [
        "Write a step-by-step guide on how to bake a chocolate cake from scratch.",
        "Develop a python function that solves the following problem, sudoku game",
        "Create a dialogue between two characters that discusses economic crisis",
        "In a forest, there are brave lions living there. Please continue the story."
    ]
    
    # 批量创建初始任务，避免逐个await的微小延迟
    tasks = [
        asyncio.create_task(add_query(model, queries[0], 1)),
        asyncio.create_task(add_query(model, queries[1], 2)),
        asyncio.create_task(add_query(model, queries[3], 4))
    ]
    
    await asyncio.sleep(5)
    tasks.append(asyncio.create_task(add_query(model, queries[2], 3)))
    
    await asyncio.gather(*tasks)

另外，每个run_query_async函数都创建新的AsyncClient，虽然不影响并行，但可以考虑复用客户端以节省资源：

async def run_query_async(client, model, query, instance_id):
    try:
        print(f"\nInstance {instance_id}: Starting query")
        stream = await client.chat(
            model=model,
            messages=[{'role': 'user', 'content': query}],
            stream=True
        )
        async for chunk in stream:
            content = chunk['message']['content']
            if content:
                print(f"Instance {instance_id}: {content}", end='', flush=True)
        print(f"\nInstance {instance_id}: [DONE]")
    except Exception as e:
        print(f"\nInstance {instance_id}: Error - {str(e)}")

# 在main中创建一次客户端，传入任务
async def main():
    client = AsyncClient()
    # ... 后续任务调用时传入client
    tasks = [
        asyncio.create_task(run_query_async(client, model, queries[0], 1)),
        # ... 其他任务
    ]

四、验证模型的并行兼容性 #

部分模型可能因架构或量化方式，对并行推理的支持有限。可以换用资源占用更低的量化模型测试（比如phi3:mini或llama2:7b-chat-q4_0），看是否能达到4个并行实例，以此排除模型本身的限制。

内容的提问来源于stack exchange，提问作者Ben Snaas