Azure AI Foundry Agents:构建智能的、目标驱动的自主 AI 系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import Agent, AgentThread

# 创建专门的研究 Agent
research_agent = client.agents.create(
    model="gpt-4",
    name="研究助手",
    instructions="你是一个专业的研究助手,擅长收集和分析信息。"
)

# 创建专门的写作 Agent
writing_agent = client.agents.create(
    model="gpt-4",
    name="写作助手",
    instructions="你是一个专业的写作助手,擅长将信息整理成文章。"
)

1
2
3
4
5
6
7

# 为 Agent 添加代码解释器工具
agent = client.agents.create(
    model="gpt-4",
    name="数据分析师",
    instructions="你是一个数据分析专家。",
    tools=[{"type": "code_interpreter"}]
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 创建新的对话线程
thread = client.agents.create_thread()

# 在线程中添加消息
message = client.agents.create_message(
    thread_id=thread.id,
    role="user",
    content="请帮我分析这个数据集"
)

# 运行 Agent
run = client.agents.create_run(
    thread_id=thread.id,
    assistant_id=agent.id
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# 上传数据文件
file = client.agents.upload_file(
    file_path="sales_data.csv",
    purpose="assistants"
)

# 创建分析 Agent
analyst = client.agents.create(
    model="gpt-4",
    name="销售数据分析师",
    instructions="分析销售数据并提供洞察。",
    tools=[{"type": "code_interpreter"}],
    tool_resources={
        "code_interpreter": {
            "file_ids": [file.id]
        }
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import time

def run_agent_with_retry(client, thread_id, agent_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            run = client.agents.create_run(
                thread_id=thread_id,
                assistant_id=agent_id
            )
            
            # 等待运行完成
            while run.status in ["queued", "in_progress"]:
                time.sleep(1)
                run = client.agents.get_run(
                    thread_id=thread_id,
                    run_id=run.id
                )
            
            if run.status == "completed":
                return run
            elif run.status == "failed":
                print(f"运行失败 (尝试 {attempt + 1}/{max_retries}): {run.last_error}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数退避
                continue
        except Exception as e:
            print(f"异常 (尝试 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
            continue
    
    raise Exception("Agent 运行失败,已达到最大重试次数")

1
2
3
4
5
6
7
8
9

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def log_agent_run(thread_id, agent_id, run):
    logger.info(f"Agent 运行: thread={thread_id}, agent={agent_id}, status={run.status}")
    if run.status == "failed":
        logger.error(f"错误详情: {run.last_error}")

1
2
3
4
5
6
7

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential

client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str="<your-connection-string>"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 创建带有文件搜索功能的 Agent
agent = client.agents.create(
    model="gpt-4",
    name="知识库助手",
    instructions="你是一个知识库助手,帮助用户查找信息。",
    tools=[{"type": "file_search"}],
    tool_resources={
        "file_search": {
            "vector_stores": [{"file_ids": file_ids}]
        }
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 定义函数工具
tools = [{
    "type": "function",
    "function": {
        "name": "get_order_status",
        "description": "获取订单状态",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "订单ID"
                }
            },
            "required": ["order_id"]
        }
    }
}]

agent = client.agents.create(
    model="gpt-4",
    name="订单助手",
    instructions="你是一个订单查询助手。",
    tools=tools
)

1
2
3
4
5
6
7
8

from azure.identity import DefaultAzureCredential

# 使用托管身份进行身份验证
credential = DefaultAzureCredential()
client = AIProjectClient.from_connection_string(
    credential=credential,
    conn_str="<your-connection-string>"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 获取运行步骤以了解 Agent 的决策过程
run_steps = client.agents.list_run_steps(
    thread_id=thread.id,
    run_id=run.id
)

for step in run_steps:
    print(f"步骤类型: {step.type}")
    if step.type == "tool_calls":
        for tool_call in step.step_details.tool_calls:
            print(f"工具: {tool_call.type}")
            print(f"输入: {tool_call}")