项目接入 AI 指南-阿里百炼版

项目接入 AI 指南

API 对比示例

DeepSeek 官方 curl 示例

curl -X POST https://api.deepseek.com/chat/completions \-H "Content-Type: application/json" \-H "Authorization: Bearer $DEEPSEEK_API_KEY" \-d '{"model": "deepseek-chat","messages": [{"role": "user","content": "Hello, how are you?"}],"temperature": 0.7,"stream": false}'

通义千问（百炼） curl 示例

curl -X POST https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions \
-H "Authorization: Bearer $DASHSCOPE_API_KEY" \
-H "Content-Type: application/json" \
-d '{"model": "qwen-plus","messages": [{"role": "system","content": "You are a helpful assistant."},{"role": "user","content": "你是谁？"}],"stream": false,"temperature": 0.7
}'

获取 API 密钥

通义千问 API 密钥获取

访问 阿里云 官方网站 注册账号，登录后在控制台创建 API Key。

Python 代码示例

先看官网在线调试

代码方式演示说明（示例1）：

import requests
import json
import timeclass QWenClient:def __init__(self, api_key):self.api_key = api_keyself.base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"def chat(self, messages, model="qwen3-coder-plus", temperature=0.7, stream=False):headers = {"Content-Type": "application/json","Authorization": f"Bearer {self.api_key}"}data = {"model": model,"messages": messages,"temperature": temperature,"stream": stream}try:response = requests.post(f"{self.base_url}/chat/completions",headers=headers,json=data)# 捕获HTTP错误response.raise_for_status()  if not stream:return response.json()else:return self.stream_response(response)    except Exception as e:return {"error": str(e), "response": response.text if 'response' in locals() else ""}def stream_response(self, response):for line in response.iter_lines():if line:decoded_line = line.decode('utf-8')if decoded_line.startswith('data: '):data = decoded_line[6:]if data != '[DONE]':yield json.loads(data)# 使用示例
client = QWenClient("api_key")# 简单对话
messages = [{"role": "system", "content": "你是一个有用的AI助手"},{"role": "user", "content": "你好，请介绍一下自己"}
]
#非流式输出
# response = client.chat(messages, temperature=0.7)
# if "error" in response:
#     print("错误：", response["error"])
# else:
#     print("AI-非流式输出 回复：", response['choices'][0]['message']['content'])#流式输出
print("AI -流式输出 回复：", end="", flush=True)
response_stream = client.chat(messages, stream=True)for chunk in response_stream:try:content = chunk['choices'][0]['delta'].get('content', '')print(content, end="", flush=True)time.sleep(0.02) except:continueprint()  # 最后换行

使用官方openai方式（示例2）

from openai import OpenAI
from dotenv import load_dotenv
import os
import time# 加载 .env 环境变量,通过pip install python-dotenv 安装并读取env文件内容
load_dotenv()# 初始化 AI 客户端（从环境变量读取，安全！）
client = OpenAI(api_key=os.getenv("ALIYUN_API_KEY"),base_url=os.getenv("ALIYUN_BASE_URL")
)# 会话记忆核心：维护 messages 列表
messages = [{"role": "system", "content": "你是一个有用的AI助手"}
]# ====================
# 【1】流式输出（逐字打印）
# ====================
user_input = "你好，请介绍一下自己，200字左右"
messages.append({"role": "user", "content": user_input})print("AI（流式）：", end="", flush=True)stream = client.chat.completions.create(model=os.getenv("DEFAULT_MODEL"),messages=messages,stream=True,temperature=0.7
)# 接收流式内容
full_response = ""
for chunk in stream:content = chunk.choices[0].delta.content or ""if content:full_response += contentprint(content, end="", flush=True)time.sleep(0.02)print()# 把AI回复加入历史（会话记忆关键步骤）
messages.append({"role": "assistant", "content": full_response})# ====================
# 【2】测试会话记忆（第二轮对话，AI 会记住历史）
# ====================
print("\n--- 第二轮对话（记忆测试）---")
user_input2 = "我刚才让你做什么了？"
messages.append({"role": "user", "content": user_input2})resp = client.chat.completions.create(model=os.getenv("DEFAULT_MODEL"),messages=messages,temperature=0.7
)ai_msg = resp.choices[0].message.content
print("AI：", ai_msg)
messages.append({"role": "assistant", "content": ai_msg})

需要在项目根目录下创建.env文件，配置内容是key=value模式:

ALIYUN_API_KEY=实际的api_key
ALIYUN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
DEFAULT_MODEL=qwen3.6-plus

关键参数详解

messages 数组中 role 的含义

role="system" 是系统提示词，用于：

• 设定AI角色：如"你是一个专业的客服助手"
• 定义行为准则：如"请保持礼貌和专业"
• 提供上下文信息：如"用户是VIP客户，请优先处理"

messages = [{"role": "system", "content": "你是Qwen3 AI助手，专门帮助用户解决技术问题"},{"role": "user", "content": "如何优化数据库查询性能？"},{"role": "assistant", "content": "首先需要分析慢查询日志..."}
]

另外，其他两个值如下：

• role=user 用户角色，就是我们自己，给ai提交问题
• assistant ai回复角色，就是ai回复的结果。

temperature 参数作用

• 低值 (0.1-0.3)：输出更确定、保守，适合任务导向场景
• 中值 (0.5-0.7)：平衡创造性与准确性
• 高值 (0.8-1.0)：更有创造性但可能不准确，适合创意生成

stream 参数

# 流式响应示例
def stream_chat():for chunk in client.chat(messages, stream=True):if 'choices' in chunk and len(chunk['choices']) > 0:delta = chunk['choices'][0].get('delta', {})content = delta.get('content', '')if content:print(content, end='', flush=True)  # 实时显示

会话记忆实现

官方多轮对话 说明，

关键点：每一轮对话都需要将用户的最新提问和模型的回复追加到message数组中，并将其作为下一次请求的输入。
示例见上述使用openai的示例（示例2）。

多模态

除了可以传递文本外，还可以把图片、音频等作为消息。详见官网，这里只演示本地图片方式

图片消息-本地图片

def image_to_base64(image_path):with open(image_path, "rb") as image_file:encoded_string = base64.b64encode(image_file.read())return f"data:image/jpeg;base64,{encoded_string.decode('utf-8')}"# 支持图片的对话格式
messages_with_image = [{"role": "user","content": [{"type": "text", "text": "请描述这张图片的内容"},{"type": "image_url", "image_url": {"url": image_to_base64("path/to/image.jpg")}}]}
]

LangChain与Agent

什么是LangChain

LangChain大模型开发框架，帮你快速把 LLM（如通义千问、GPT）、工具、数据库、记忆、文档等组装成 AI 应用。

什么是Agent

Agent（智能体）：一种能自主思考、做计划、调用工具、完成任务的 LLM 工作模式。 智能体= LLM + 工具调用能力 + 记忆 + 规划执行。简单来说，给它发送一个指令，它能自动完成特定的目标，而不是局限于聊天机器人的一问一答。

区别

Agent 是一种功能，LangChain 是实现 Agent 的最常用框架。

智能体 Agent 使用方式

安装LangChain

pip install langchain

LangChain ji集成openai

pip install langchain-openai

简单 Agent 实现：查询出游计划（模拟天气和火车查询，示例3）

from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage# 加载环境变量
load_dotenv()# 初始化 LLM
llm = ChatOpenAI(api_key=os.getenv("ALIYUN_API_KEY"),base_url=os.getenv("ALIYUN_BASE_URL"),model=os.getenv("DEFAULT_MODEL"),temperature=0.1
)# ================== 工具 ==================
@tool
def get_weather(city: str):"""查询城市天气"""return f"{city}：晴天 25℃，适合旅行"@tool
def get_train(departure: str, destination: str):"""查询火车信息"""return f"{departure} → {destination}：高铁G102，08:00发车，12:30到达"tools = [get_weather, get_train]agent = create_agent(llm, tools)# 执行
inputs = [HumanMessage(content="帮我规划北京去上海的2天旅行计划，需要查天气和火车")]
res = agent.invoke({"messages": inputs})# 输出结果
print("\n===== 最终旅游计划 =====")
print(res["messages"][-1].content)

消息对象Message

langChain也对消息做了对象封装为BaseMessage，其子类如下：

消息类型	对应 Role
SystemMessage	system
HumanMessage	user
AIMessage	assistant
ToolMessage	tool

那么对于不同的消息，使用不同的对象即可，类似java的面向对象。
对上述的示例2进行改造，看起来会简洁很多（示例4）：

from dotenv import load_dotenv
import os
import time
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessageload_dotenv()llm = ChatOpenAI(api_key=os.getenv("ALIYUN_API_KEY"),base_url=os.getenv("ALIYUN_BASE_URL"),model=os.getenv("DEFAULT_MODEL"),temperature=0.7
)# ✅ 省略 content=，直接传字符串
messages = [SystemMessage("你是一个有用的AI助手")  
]# 【1】流式输出
user_input = "你好，请介绍一下自己，200字左右"
messages.append(HumanMessage(user_input))  print("AI（流式）：", end="", flush=True)
stream = llm.stream(messages)full_response = ""
for chunk in stream:content = chunk.content or ""if content:full_response += contentprint(content, end="", flush=True)time.sleep(0.02)print()
messages.append(AIMessage(full_response))  # 【2】测试会话记忆
print("\n--- 第二轮对话（记忆测试）---")
user_input2 = "我刚才让你做什么了？"
messages.append(HumanMessage(user_input2))  resp = llm.invoke(messages)
ai_msg = resp.content
print("AI：", ai_msg)
messages.append(AIMessage(ai_msg))  

注：自定义的工具的描述很重要（方法注释），需要告诉智能体这个方法是做什么。

预定义工具 Tavily

除了自定义工具外，Tavily有一些联网的工具可以直接使用，Tavily文档说明

• 安装

pip install tavily-python

• 获取api_key，注册且登录官网后默认有一个api_key供使用
• 引入并使用（示例5）

from tavily import TavilyClient
from dotenv import load_dotenv
import osload_dotenv()tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
response = tavily_client.search("今天武汉的天气如何")print(response)

这是直接使用客户端，那么langchain也支持直接使用

• 安装

pip install langchain-tavily

对上述天气的示例（示例4）进行改造

from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from tavily import TavilyClient# 加载环境变量
load_dotenv()# 初始化 LLM
llm = ChatOpenAI(api_key=os.getenv("ALIYUN_API_KEY"),base_url=os.getenv("ALIYUN_BASE_URL"),model=os.getenv("DEFAULT_MODEL"),temperature=0.1
)
tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
# ================== 工具 ==================
@tool
def get_weather(city: str):"""查询城市天气"""return tavily_client.search(city)@tool
def get_train(departure: str, destination: str):"""查询火车信息"""return f"{departure} → {destination}：高铁G102，08:00发车，12:30到达"tools = [get_weather, get_train]agent = create_agent(llm, tools)# 执行
inputs = [HumanMessage(content="帮我规划北京去上海的2天旅行计划，需要查天气和火车")]
res = agent.invoke({"messages": inputs})# 输出结果
print("\n===== 最终旅游计划 =====")
print(res["messages"][-1].content)

短期记忆

LangChain的短期记忆（即会话层面），通过AgentState来实现的。
LangChain提供了checkpointer对象来保存AgentState，每一次用户与AI的交互都会生成一个快照，记录为一个checkpoint（快照）。每次会话都对应一个thread_id。
在上述示例4中，对于历史的消息都是手动传入到消息的，其实LangGraph有更好的方法。（示例7）

from dotenv import load_dotenv
import os
import time
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver  # 短期记忆load_dotenv()llm = ChatOpenAI(api_key=os.getenv("ALIYUN_API_KEY"),base_url=os.getenv("ALIYUN_BASE_URL"),model=os.getenv("DEFAULT_MODEL"),temperature=0.7
)# 使用create_agent创建智能体
agent = create_agent(model=llm,tools=[], checkpointer=InMemorySaver()  # 绑定短期记忆
)# 会话ID：区分不同用户的记忆（关键！）
config = {"configurable": {"thread_id": "zxh_001"}}initial_messages = [SystemMessage("你是一个有用的AI助手")  
]# 【1】流式输出（保留原打印格式、逐字效果）
user_input = "你好，请介绍一下自己，200字左右"
initial_messages.append(HumanMessage(user_input))  print("AI（流式）：", end="", flush=True)
# 替换原llm.stream为agent.stream，带记忆调用
stream = agent.stream({"messages": initial_messages},config=config,stream_mode="values"  # 保证流式输出格式统一
)full_response = ""
for chunk in stream:# 提取最新消息内容（避免重复打印）msg = chunk["messages"][-1]content = msg.content or ""if content and content not in full_response:new_content = content[len(full_response):]full_response = contentprint(new_content, end="", flush=True)time.sleep(0.02)print()# 【2】测试会话记忆（保留原格式，无需手动维护messages）
print("\n--- 第二轮对话（记忆测试）---")
user_input2 = "我刚才让你做什么了？"
# 仅传入新的用户消息，记忆自动加载
second_messages = [HumanMessage(user_input2)]  # 替换原llm.invoke为agent.invoke，带记忆调用
resp = agent.invoke({"messages": second_messages},config=config  # 复用会话ID，读取历史记忆
)ai_msg = resp["messages"][-1].content
print("AI：", ai_msg)

此时就可以在会话中实现历史记忆，但此时是存储在内存中，当然也支持讲会话数据保存到数据库。