协作头脑风暴
brainstorming
在实现前澄清需求、探索设计选择,并形成清晰的开发方向。
azure-ai-agents-py【开发与工程】Build AI agents using the Azure AI Agents Python SDK (azure-ai-agents). Use when creating agents hosted on Azure AI Foundry with tools (File Search, Code Interpreter, Bing Grounding, Azure AI Search, Function Calling, OpenAPI, MCP), managing threads and messages, implementing streaming responses, or working with vector stores. This is the low-level S…
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
Build agents hosted on Azure AI Foundry using the azure-ai-agents SDK.
pip install azure-ai-agents azure-identity
# Or with azure-ai-projects for additional features
pip install azure-ai-projects azure-identity
PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"
MODEL_DEPLOYMENT_NAME="gpt-4o-mini"
from azure.identity import DefaultAzureCredential
from azure.ai.agents import AgentsClient
credential = DefaultAzureCredential()
client = AgentsClient(
endpoint=os.environ["PROJECT_ENDPOINT"],
credential=credential,
)
The basic agent lifecycle: create agent → create thread → create message → create run → get response
import os
from azure.identity import DefaultAzureCredential
from azure.ai.agents import AgentsClient
client = AgentsClient(
endpoint=os.environ["PROJECT_ENDPOINT"],
credential=DefaultAzureCredential(),
)
# 1. Create agent
agent = client.create_agent(
model=os.environ["MODEL_DEPLOYMENT_NAME"],
name="my-agent",
instructions="You are a helpful assistant.",
)
# 2. Create thread
thread = client.threads.create()
# 3. Add message
client.messages.create(
thread_id=thread.id,
role="user",
content="Hello!",
)
# 4. Create and process run
run = client.runs.create_and_process(thread_id=thread.id, agent_id=agent.id)
# 5. Get response
if run.status == "completed":
messages = client.messages.list(thread_id=thread.id)
for msg in messages:
if msg.role == "assistant":
print(msg.content[0].text.value)
# Cleanup
client.delete_agent(agent.id)
| Tool | Class | Use Case |
|---|---|---|
| Code Interpreter | CodeInterpreterTool | Execute Python, generate files |
| File Search | FileSearchTool | RAG over uploaded documents |
| Bing Grounding | BingGroundingTool | Web search |
| Azure AI Search | AzureAISearchTool | Search your indexes |
| Function Calling | FunctionTool | Call your Python functions |
| OpenAPI | OpenApiTool | Call REST APIs |
| MCP | McpTool | Model Context Protocol servers |
See references/tools.md for detailed patterns.
from azure.ai.agents import CodeInterpreterTool, FileSearchTool
agent = client.create_agent(
model=os.environ["MODEL_DEPLOYMENT_NAME"],
name="tool-agent",
instructions="You can execute code and search files.",
tools=[CodeInterpreterTool()],
tool_resources={"code_interpreter": {"file_ids": [file.id]}},
)
from azure.ai.agents import FunctionTool, ToolSet
def get_weather(location: str) -> str:
"""Get weather for a location."""
return f"Weather in {location}: 72F, sunny"
functions = FunctionTool(functions=[get_weather])
toolset = ToolSet()
toolset.add(functions)
agent = client.create_agent(
model=os.environ["MODEL_DEPLOYMENT_NAME"],
name="function-agent",
instructions="Help with weather queries.",
toolset=toolset,
)
# Process run - toolset auto-executes functions
run = client.runs.create_and_process(
thread_id=thread.id,
agent_id=agent.id,
toolset=toolset, # Pass toolset for auto-execution
)
from azure.ai.agents import AgentEventHandler
class MyHandler(AgentEventHandler):
def on_message_delta(self, delta):
if delta.text:
print(delta.text.value, end="", flush=True)
def on_error(self, data):
print(f"Error: {data}")
with client.runs.stream(
thread_id=thread.id,
agent_id=agent.id,
event_handler=MyHandler(),
) as stream:
stream.until_done()
See references/streaming.md for advanced patterns.
file = client.files.upload_and_poll(
file_path="data.csv",
purpose="assistants",
)
vector_store = client.vector_stores.create_and_poll(
file_ids=[file.id],
name="my-store",
)
agent = client.create_agent(
model=os.environ["MODEL_DEPLOYMENT_NAME"],
tools=[FileSearchTool()],
tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)
from azure.ai.agents.aio import AgentsClient
async with AgentsClient(
endpoint=os.environ["PROJECT_ENDPOINT"],
credential=DefaultAzureCredential(),
) as client:
agent = await client.create_agent(...)
# ... async operations
See references/async-patterns.md for async patterns.
agent = client.create_agent(
model=os.environ["MODEL_DEPLOYMENT_NAME"],
response_format={"type": "json_object"},
)
agent = client.create_agent(
model=os.environ["MODEL_DEPLOYMENT_NAME"],
response_format={
"type": "json_schema",
"json_schema": {
"name": "weather_response",
"schema": {
"type": "object",
"properties": {
"temperature": {"type": "number"},
"conditions": {"type": "string"},
},
"required": ["temperature", "conditions"],
},
},
},
)
# Save thread_id for later
thread_id = thread.id
# Resume later
client.messages.create(
thread_id=thread_id,
role="user",
content="Follow-up question",
)
run = client.runs.create_and_process(thread_id=thread_id, agent_id=agent.id)
messages = client.messages.list(thread_id=thread.id, order="asc")
for msg in messages:
role = msg.role
content = msg.content[0].text.value
print(f"{role}: {content}")
client.delete_agent(agent.id)create_and_process for simple cases, streaming for real-time UX*_and_poll methods for long operations