阿里云百炼模型入门篇-大语言模型

from http import HTTPStatus
from dashscope import Generation
def call_with_stream():
    messages = [
        {'role': 'user', 'content': '如何做西红柿炖牛腩？'}]
    responses = Generation.call("qwen-max",
                                messages=messages,
                                result_format='message',  # 设置输出为'message'格式
                                stream=True, # 设置输出方式为流式输出
                                incremental_output=True  # 增量式流式输出
                                )
    for response in responses:
        if response.status_code == HTTPStatus.OK:
            print(response.output.choices[0]['message']['content'],end='')
        else:
            print('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
                response.request_id, response.status_code,
                response.code, response.message
            ))
if __name__ == '__main__':
    call_with_stream()

from http import HTTPStatus
import dashscope
def send_message(user_message: str) -> str:
    global messages
    messages.append({'role': 'user', 'content': user_message})
    response = dashscope.Generation.call(model="qwen-turbo",
                                         messages=messages,
                                         # 将输出设置为"message"格式
                                         result_format='message')
    if response.status_code == HTTPStatus.OK:
        # 将assistant的回复添加到messages列表中
        messages.append({'role': response.output.choices[0]['message']['role'],
                         'content': response.output.choices[0]['message']['content']})
        return response.output.choices[0]['message']['content']
    else:
        # 如果响应失败，将最后一条user message从messages列表里删除，确保user/assistant消息交替出现
        messages = messages[:-1]
        return response.message
if __name__ == '__main__':
    messages = [
        {'role': 'system', 'content': '你是一个旅游助手.'}
    ]
    # 第一轮对话
    result = send_message("我想去新疆")
    print("第一轮结果: \n%s" % result)
    # 第二轮对话
    result = send_message("哪里有什么特色美食")
    print("第二轮结果: \n%s" % result)

import json
import random
from dashscope import Generation
# 定义工具列表，模型在选择使用哪个工具时会参考工具的name和description
tools = [
    # 工具1 获取指定城市的天气
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "当你想查询指定城市的天气时非常有用。",
            "parameters": {  # 查询天气时需要提供位置，因此参数设置为location
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "城市或县区，比如北京市、杭州市、余杭区等。"
                    }
                }
            },
            "required": [
                "location"
            ]
        }
    }
]
# 模拟天气查询工具。返回结果示例：“北京今天是晴天。”
def get_current_weather(location):
    return f"{location}今天是晴天。 "
# 封装模型响应函数
def get_response(messages):
    response = Generation.call(
        model='qwen-max',
        messages=messages,
        tools=tools,
        seed=random.randint(1, 10000),  # 设置随机数种子seed，如果没有设置，则随机数种子默认为1234
        result_format='message'  # 将输出设置为message形式
    )
    return response
def call_with_messages(message: str):
    print('\n')
    messages = [
        {
            "content": message,
            "role": "user"
        }
    ]
    # 模型的第一轮调用
    first_response = get_response(messages)
    assistant_output = first_response.output.choices[0].message
    # print(f"\n大模型第一轮输出信息：{first_response}\n")
    messages.append(assistant_output)
    if 'tool_calls' not in assistant_output:  # 如果模型判断无需调用工具，则将assistant的回复直接打印出来，无需进行模型的第二轮调用
        print(f"最终答案：{assistant_output.content}")  # 此处直接返回模型的回复，您可以根据您的业务，选择当无需调用工具时最终回复的内容
        return
    # 如果模型选择的工具是get_current_weather
    elif assistant_output.tool_calls[0]['function']['name'] == 'get_current_weather':
        tool_info = {"name": "get_current_weather", "role": "tool"}
        location = json.loads(assistant_output.tool_calls[0]['function']['arguments'])['properties']['location']
        tool_info['content'] = get_current_weather(location)
    print(f"工具输出信息：{tool_info['content']}\n")
    messages.append(tool_info)
    # 模型的第二轮调用，对工具的输出进行总结
    second_response = get_response(messages)
    # print(f"大模型第二轮输出信息：{second_response}\n")
    print(f"最终答案：{second_response.output.choices[0].message['content']}")
if __name__ == '__main__':
    call_with_messages("今天北京的天气如何?")

import json
import logging
import traceback
from http import HTTPStatus
from typing import Generator
from dashscope import Generation
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def create_app():
    """Create the FastAPI app and include the router."""
    fa_app = FastAPI()
    origins = [
        "*",
    ]
    fa_app.add_middleware(
        CORSMiddleware,
        allow_origins=origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    return fa_app
# 创建fastapi app
app = create_app()
# 定义后端api的输入输出参数结构
class ChatRequest(BaseModel):
    message: str = Field(default=None, alias='message')
class ChatResponse(BaseModel):
    content: str = Field(default=None, alias='content')
messages = [
    {'role': 'system', 'content': '你是一个旅游助手.'}
]
def sse_data(data: str) -> str:
    return f"data: {data}\n\n"
def chat_generator(request: ChatRequest) -> Generator[str, None, None]:
    try:
        global messages
        message = request.message
        messages.append({'role': 'user', 'content': message})
        logging.info("send message to llm: %s" % (json.dumps(messages, ensure_ascii=False)))
        responses = Generation.call("qwen-max",
                                    messages=messages,
                                    result_format='message',
                                    stream=True,
                                    incremental_output=True
                                    )
        full_message = ""
        for response in responses:
            if response.status_code == HTTPStatus.OK:
                data = response.output.choices[0]['message']['content']
                full_message += data
            else:
                logging.error('Request id: %s, Status code: %s, error code: %s, error message: %s' % (
                    response.request_id, response.status_code, response.code, response.message
                ))
                data = response.message
            result = ChatResponse(content=data)
            yield sse_data(json.dumps(result.dict(), ensure_ascii=False))
        if full_message is not None and full_message:
            messages.append({'role': 'assistant', 'content': full_message})
    except Exception as e:
        logging.error("failed to do stream chat, request: %s, error: %s" % (request, traceback.format_exc()))
        yield sse_data(str(e))
@app.post("/v1/chat", summary="chat api")
async def chat(request: ChatRequest):
    logger.info("start chat, request: %s" % json.dumps(request.dict(), ensure_ascii=False))
    return StreamingResponse(chat_generator(request=request), media_type="text/event-stream")
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8080)