应用引入LLM实践-一次性输出和流式输出(思维链)

在大模型应用时，有的场景希望根据prompt要求一次性输出结果，有的场景则希望输出整个思维过程以及最后的结果。
这部分在网上看了一些文章说的都不一样，自己尝试了一下，正确的写法是这样的，记录一下。
一次性输出：

1. from openai import OpenAI
3. def generate_huoshan(prompt):
4.     client = OpenAI(
5.     # 从环境变量中读取您的方舟API Key
6.         api_key="**",
7.         base_url="https://ark.cn-beijing.volces.com/api/v3",
8.         # 深度推理模型耗费时间会较长，建议您设置一个较长的超时时间，推荐为30分钟
9.         timeout=1800,
10.     )
12.     response = client.chat.completions.create(
13.         model="deepseek-r1-250120",
14.         messages=[
15.             {"role": "system", "content": "You are a professional market research assistant who needs to accurately obtain retail price information for specified electronic products in a specific market"},
16.             {"role": "user", "content": prompt},
18.         ],
19.         max_tokens=1024,
20.         temperature=0.6,
21.         stream=False
22.     )
23.     answer = response.choices[0].message.content
24.     return answer.strip()

复制代码

View Code流式输出：

1. from openai import OpenAI
3. def generate_huoshan(prompt):
4.     client = OpenAI(
5.         api_key="*",
6.         base_url="https://ark.cn-beijing.volces.com/api/v3",
7.         # 深度推理模型耗费时间会较长，建议您设置一个较长的超时时间，推荐为30分钟
8.         timeout=1800,
9.     )
11.     response = client.chat.completions.create(
12.     model="deepseek-r1-250120",
13.     messages=[
14.     {"role": "system", "content": "You are a professional market research assistant who needs to accurately obtain retail price information for specified electronic products in a specific market"},
15.     {"role": "user", "content": prompt},
17.     ],
18.     max_tokens=1024,
19.     temperature=0.6,
20.     stream=True
21.     )
23.     for chunk in response:
24.         delta = chunk.choices[0].delta
25.         # 优先提取思维链内容
26.         if hasattr(delta, 'reasoning_content') and delta.reasoning_content:
27.             yield delta.reasoning_content
28.             #print(f"[推理过程] {delta.reasoning_content}", end="\n", flush=True)
29.         # 处理最终回答内容
30.         elif delta.content:
31.             yield delta.content
32.             #print(f"[最终回答] {delta.content}", end="", flush=True)
33.         else:
34.             continue

复制代码

View Code外层通过这样返回：

1.    def generate_stream():
2.         try:
3.             for chunk in generate_text(model_name, prompt):
4.                 #yield chunk
5.                 yield json.dumps({"msg": "Success", "code": 200, "data": chunk})+ '\n'
6.                
7.         except Exception as e:
8.                 yield json.dumps({"code": 500, "message": str(e)})+ '\n'  # Yield a JSON string
9.     headers = {
10.         'Content-Type':'text/event-stream',
11.         'Cache-Control': 'no-cache',
12.         'X-Accel-Buffering':'no',
13.     }
14.     return Response(generate_stream(), mimetype='text/event-stream',headers=headers)

复制代码

View Code然后，前端相应做解析即可。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！