本文是AI Agent 30天速成系列的第一天学习笔记，聚焦大模型基础核心概念、异步LLM调用封装、提示工程核心技巧与SSE流式接口开发，所有代码可直接运行，适合AI Agent开发入门学习者。
| 参数 | 作用 | 业务取值建议 |
|---|---|---|
| temperature | 控制输出随机性，0=完全确定，1=自由发散 | 知识库/工具调用：0.00.1；文案创作：0.70.9 |
| top_p | 核采样，控制候选词范围 | 工具场景0.1，创作场景0.8 |
| max_tokens | 限制单次返回最大输出长度 | 工具查询设512，长文档总结设2048 |
国内主流模型均兼容OpenAI v1接口格式，仅需更换base_url和api_key，即可用一套代码对接所有模型。
data:分隔流式返回块，过滤空行、[DONE]结束标记import aiohttp
import asyncio
import re
from pydantic import BaseModel, Field
from typing import Optional, AsyncGenerator
# 模型配置映射
MODEL_CONFIG = {
"qwen-turbo": {
"base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
"api_key": "你的通义千问key"
},
"deepseek-chat": {
"base_url": "https://api.deepseek.com/v1/chat/completions",
"api_key": "你的deepseek key"
}
}
# 结构化输出校验模型示例
class OrderQueryResp(BaseModel):
order_name: str = Field(description="订单名称")
order_price: float = Field(description="订单价格")
delivery_status: str = Field(description="配送状态")
class AsyncLLMClient:
def __init__(self, model_name: str = "qwen-turbo"):
self.model = model_name
self.conf = MODEL_CONFIG[model_name]
self.semaphore = asyncio.Semaphore(5) # 最大并发5
async def _request(self, payload: dict, stream: bool = False):
headers = {
"Authorization": f"Bearer {self.conf['api_key']}",
"Content-Type": "application/json"
}
timeout = aiohttp.ClientTimeout(total=60)
async with self.semaphore:
async with aiohttp.ClientSession(timeout=timeout) as session:
try:
async with session.post(self.conf["base_url"], json=payload, headers=headers) as resp:
if resp.status != 200:
err = await resp.text()
raise Exception(f"模型接口异常: {resp.status}, {err}")
if stream:
return self._stream_parse(resp)
return await resp.json()
except aiohttp.ClientError as e:
raise Exception(f"网络请求失败: {str(e)}")
async def _stream_parse(self, response) -> AsyncGenerator[str, None]:
buffer = ""
async for chunk in response.content.iter_chunked(1024):
buffer += chunk.decode("utf-8")
# 分割流式块
while "data:" in buffer:
idx = buffer.find("data:")
end_idx = buffer.find("\n\n", idx)
if end_idx == -1:
break
data_block = buffer[idx+5:end_idx].strip()
buffer = buffer[end_idx+2:]
if data_block == "[DONE]":
return
try:
import json
data = json.loads(data_block)
content = data["choices"][0]["delta"].get("content", "")
if content:
yield content
except:
continue
async def chat_sync(self, prompt: str, temperature: float = 0.1) -> str:
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"stream": False
}
res = await self._request(payload)
return res["choices"][0]["message"]["content"]
async def chat_stream(self, prompt: str, temperature: float = 0.1):
payload = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"temperature": temperature,
"stream": True
}
async for text in await self._request(payload, stream=True):
yield text
# 强制JSON输出封装
async def chat_json(self, prompt: str, resp_model: BaseModel) -> BaseModel:
json_prompt = f"""
{prompt}
严格要求：仅返回标准JSON，禁止任何解释、注释、markdown、多余文字。
JSON字段要求：{resp_model.model_json_schema()}
"""
raw = await self.chat_sync(json_prompt, temperature=0.0)
# 正则提取JSON
match = re.search(r"\{.*\}", raw, re.S)
if not match:
# 解析失败重试一次
raw = await self.chat_sync(json_prompt, temperature=0.0)
match = re.search(r"\{.*\}", raw, re.S)
if not match:
raise Exception("模型无法输出合法JSON")
json_str = match.group()
return resp_model.model_validate_json(json_str)
# 测试入口
async def test_client():
client = AsyncLLMClient("qwen-turbo")
# 测试1：同步JSON结构化输出
res = await client.chat_json(
prompt="模拟一个手机订单信息",
resp_model=OrderQueryResp
)
print("结构化结果：", res.model_dump())
# 测试2：流式输出
print("\n流式输出：")
async for chunk in client.chat_stream("简单介绍AI Agent"):
print(chunk, end="")
if __name__ == "__main__":
asyncio.run(test_client())
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
from llm_client import AsyncLLMClient
app = FastAPI(title="Day1 LLM流式服务")
llm = AsyncLLMClient()
# SSE封装生成器
async def stream_generator(user_prompt: str):
yield "data: 开始生成回答\n\n"
async for text in llm.chat_stream(user_prompt):
yield f"data: {text}\n\n"
yield "data: [DONE]\n\n"
@app.get("/chat/stream")
async def chat_stream(prompt: str):
return StreamingResponse(
stream_generator(prompt),
media_type="text/event-stream"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", reload=True)
llm_client.py测试JSON结构化输出，观察Pydantic校验效果http://127.0.0.1:8000/docs调试流式接口temperature参数（0/0.9），对比模型输出随机性差异Token是大模型处理文本的最小单位（中文1字≈2token，英文1单词≈1token），是计费和上下文限制的核心单位。
上下文窗口超限后果：输入被截断导致上下文丢失、多轮对话失忆；输出不完整、逻辑断裂；接口返回400错误；长文本生成任务提前终止。
temperature控制输出随机性（0~2），值越低输出越确定，值越高越发散。
工具调用要求参数精准，高随机性会导致参数非法、格式错乱，0~0.1接近贪心采样，能最大化结构化参数准确率。
OpenAI是行业事实标准，兼容后可一套代码对接多模型、复用生态工具链、降低学习成本、灵活选型对比效果。
response_format=json_object（OpenAI）或厂商对应参数temperature=0降低随机性data:拆分完整块后解析| 维度 | requests（同步） | aiohttp（异步） |
|---|---|---|
| 并发模型 | 多线程/多进程，阻塞IO，切换开销大 | 单线程协程，非阻塞IO，切换开销极低 |
| 并发上限 | 线程数受限，高并发资源占用高 | 单线程支撑上千并发，资源占用低 |
| 开发成本 | 逻辑简单，调试容易 | 需async/await改造，调试复杂度高 |
| 适用场景 | 低并发脚本、简单单次调用 | 高并发批量调用、服务端后端、Agent多工具并行 |
Day1重点掌握大模型基础概念、异步LLM封装、结构化输出与SSE流式接口开发，这些是AI Agent开发的核心底层能力。后续将基于此拓展多轮对话、记忆模块、工具调用等进阶内容，建议反复调试代码，吃透面试高频考点。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。