67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
智能体模块
|
||
"""
|
||
|
||
# 列举导入模块
|
||
from typing import List
|
||
|
||
from pydantic_ai import Agent, AgentRunResult, ModelMessage
|
||
from pydantic_ai.models.openai import OpenAIChatModel
|
||
from pydantic_ai.output import OutputSpec
|
||
from pydantic_ai.providers.openai import OpenAIProvider
|
||
|
||
|
||
class BaseAgent:
|
||
"""
|
||
通用智能体基类，支持：
|
||
1）实例智能体
|
||
2）异步运行
|
||
"""
|
||
|
||
def __init__(self, instructions: str, output_type: OutputSpec = str):
|
||
"""
|
||
初始化智能体
|
||
:param instructions: 指令
|
||
:param output_type: 输出类型
|
||
:return: 智能体实例
|
||
"""
|
||
# 实例智能体
|
||
self.agent = self._instantiate(
|
||
instructions=instructions,
|
||
output_type=output_type,
|
||
)
|
||
|
||
def _instantiate(self, instructions: str, output_type: OutputSpec) -> Agent:
|
||
"""
|
||
实例智能体
|
||
:param instructions: 指令
|
||
:param output_type: 输出类型
|
||
:return: 智能体实例
|
||
"""
|
||
agent = Agent(
|
||
model=OpenAIChatModel(
|
||
model_name="deepseek-v4-flash",
|
||
provider=OpenAIProvider(
|
||
base_url="https://tokenhub.tencentmaas.com/v1",
|
||
api_key="sk-D9Y1mCe8VlvNqLuSC4mAjqEwxJ2nW4C0h8a7EPn8kg9RLsHq",
|
||
),
|
||
),
|
||
instructions=instructions,
|
||
output_type=output_type,
|
||
)
|
||
return agent
|
||
|
||
async def run(
|
||
self, user_prompt: str | List[str], message_history: List[ModelMessage] = []
|
||
) -> AgentRunResult:
|
||
"""
|
||
异步运行
|
||
:param user_prompt: 用户提示词
|
||
:param message_history: 历史消息
|
||
:return: 智能体回复
|
||
"""
|
||
return await self.agent.run(
|
||
user_prompt=user_prompt, message_history=message_history
|
||
)
|