This repository AgentScope-Bricks will be merged into agentscope-runtime. Future updates, bug fixes, and maintenance will happen in the new repository. Please visit the new repository for the latest features and support.
Agentscope-Bricks is a componentized AI application development framework that provides enterprise-grade, modular components for building sophisticated AI-powered applications. Built around Alibaba Cloud Modelstudio services, it offers an unified API for creating scalable AI agents with production-ready capabilities.
- 🧩 Modular Component Architecture: Single-responsibility components with clear input/output definitions
- 🔌 Open-Source Framework Integration: Native support for LangGraph, AutoGen, LlamaIndex, and AgentScope, more on the road
- 🔒 ModelStudio Integration: Seamlessly integrate core ModelStudio enterprise components: Search, RAG, AIGC, and Memory Management.
- Getting Started
- Agent Development, Debugging & Deployment
- Why Agentscope-Bricks?
- Contributing
- License
# Basic installation
pip install agentscope-bricks
# Development installation
pip install -e .Prerequisites:
- Python 3.10+
DASHSCOPE_API_KEYenvironment variable for Alibaba Cloud services
Build type-safe, reusable components with automatic schema generation:
import asyncio
from agentscope_bricks.base.component import Component
from pydantic import BaseModel, Field
from typing import Any
# Define input/output schemas
class SearchInput(BaseModel):
query: str = Field(..., description="Search query")
max_results: int = Field(default=10, description="Maximum results")
context: dict = Field(default_factory=dict)
class SearchOutput(BaseModel):
results: list[str] = Field(..., description="Search results")
total_found: int = Field(..., description="Total results found")
context: dict = Field(..., description="Updated context")
# Implement your component
class SearchComponent(Component[SearchInput, SearchOutput]):
async def _arun(self, args: SearchInput, **kwargs: Any) -> SearchOutput:
# Your search logic here
results = [f"Result for '{args.query}' #{i+1}" for i in range(args.max_results)]
return SearchOutput(
results=results,
total_found=len(results),
context={**args.context, "last_query": args.query}
)
# Use the component
async def main():
component = SearchComponent(
name="Enterprise Search",
description="High-performance search component"
)
result = await component.arun(SearchInput(query="AI agents", max_results=5))
print(f"Found {result.total_found} results:")
for r in result.results:
print(f" - {r}")
# Component automatically generates OpenAI-compatible function schema
print("\nFunction Schema:")
print(component.function_schema)
asyncio.run(main())Build agent with Agentscope-Runtime and AgentScope
import asyncio
import os
from typing import Any
from agentscope_runtime.engine import Runner
from agentscope_runtime.engine.agents.agentscope_agent import AgentScopeAgent
from agentscope_runtime.engine.services.environment_manager import \
EnvironmentManager
from agentscope_runtime.engine.services.context_manager import \
ContextManager
from agentscope_runtime.engine.schemas.agent_schemas import (
MessageType,
RunStatus,
AgentRequest,
)
from agentscope.agent import ReActAgent
from agentscope.model import DashScopeChatModel
from agentscope_bricks.adapters.agentscope_runtime.tool import (
AgentScopeRuntimeToolAdapter,
)
from agentscope_bricks.components import Component
from pydantic import BaseModel, Field
# Initialize the instance
model = DashScopeChatModel(
"qwen-max",
api_key=os.getenv("DASHSCOPE_API_KEY"),
)
SYSTEM_PROMPT = """You are a helpful AI assistant.
# Objective
Your goal is to complete given tasks by using registry tools
"""
print("✅系统提示词已配置")
search_tool = AgentScopeRuntimeToolAdapter(SearchComponent())
# Create the AgentScope agent
agent = AgentScopeAgent(
name="Friday",
model=model,
agent_config={
"sys_prompt": SYSTEM_PROMPT,
},
tools=[search_tool],
agent_builder=ReActAgent,
)
runner_instance = Runner(
agent=agent,
context_manager=ContextManager(),
environment_manager=EnvironmentManager(),
)
# interact with agent
async def interact(runner):
# Create a request
request = AgentRequest(
input=[
{
"role": "user",
"content": [
{
"type": "text",
"text": "What is weather in beijing?",
},
],
},
],
)
# Stream interaction with the agent
print("🤖 智能体正在处理您的请求...")
async for message in runner.stream_query(
request=request,
):
# Check if this is a completed message
if (
message.object == "message"
and MessageType.MESSAGE == message.type
and RunStatus.Completed == message.status
):
all_result = message.content[0].text
print("📝 智能体输出:", all_result)
asyncio.run(interact(runner_instance))Apply component to an exists Langgraph project
# -*- coding: utf-8 -*-
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph
from pydantic import BaseModel, Field
from typing import Literal, Any
import os
from agentscope_bricks.adapters.langgraph.tool import LanggraphNodeAdapter
import asyncio
tool_node = LanggraphNodeAdapter(
[
SearchComponent(),
],
)
api_key = os.getenv("DASHSCOPE_API_KEY")
model = ChatOpenAI(
model="qwen-turbo",
openai_api_key=api_key,
openai_api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
).bind_tools(tool_node.tool_schemas)
# Define the function that determines whether to continue or not
def should_continue(state: MessagesState) -> Literal["tools", END]:
messages = state["messages"]
last_message = messages[-1]
# If the LLM makes a tool call, then we route to the "tools" node
if last_message.tool_calls:
return "tools"
# Otherwise, we stop (reply to the user)
return END
# Define the function that calls the model
def call_model(state: MessagesState):
messages = state["messages"]
response = model.invoke(messages)
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define a new graph
workflow = StateGraph(MessagesState)
# Define the two nodes we will cycle between
workflow.add_node("agents", call_model)
workflow.add_node("tools", tool_node)
# Set the entrypoint as `agents`
# This means that this node is the first one called
workflow.add_edge(START, "agents")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agents`.
# This means these are the edges taken after the `agents` node is called.
"agents",
# Next, we pass in the function that will determine which
# node is called next.
should_continue,
)
# We now add a normal edge from `tools` to `agents`.
# This means that after `tools` is called, `agents` node is called next.
workflow.add_edge("tools", "agents")
# Initialize memory to persist state between graph runs
checkpointer = MemorySaver()
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable.
# Note that we're (optionally) passing the memory when compiling the graph
app = workflow.compile(checkpointer=checkpointer)
# Use the Runnable
final_state = app.invoke(
{"messages": [HumanMessage(content="what is the weather in sf")]},
config={"configurable": {"thread_id": 42}},
)
print(final_state["messages"][-1].content)
async def arun():
final_state_async = await app.ainvoke(
{"messages": [HumanMessage(content="what is the weather in sf")]},
config={"configurable": {"thread_id": 42}},
)
print(final_state_async["messages"][-1].content)
asyncio.run(arun())- Enterprise-grade components used in production by Alibaba Cloud
- Built-in observability, tracing, and error handling
- Comprehensive testing and validation
# Works with your favorite frameworks
from agentscope_bricks.adapters.agentscope.tool import agentscope_tool_adapter
from agentscope_bricks.adapters.autogen.tool import AutogenToolAdapter
# Convert to LangGraph node
node = agentscope_tool_adapter(your_component)
# Convert to AutoGen tool
tool = AutogenToolAdapter(your_component)- Async-first architecture with streaming support
- Efficient memory management and caching
- Optimized for high-throughput scenarios
- Single-responsibility components with clear input output schemas
- Swappable implementations
We welcome contributions! Please see our development guidelines:
# Clone the repository
git clone https://github.com/your-org/agentscope-bricks.git
cd agentscope_bricks
# Install development dependencies
pip install . -U# Run all tests
pytestpre-commit install
pre-commit run --all-files- Component Development: Follow the single-responsibility principle
- Type Safety: Use Pydantic models for all inputs/outputs
- Testing: Write comprehensive unit and integration tests
- Documentation: Update relevant docs and examples
- Performance: Maintain async-first patterns
- Fork the repository
- Create a feature branch
- Make your changes with tests
- Ensure all tests pass
- Submit a pull request
This framework is licensed under the Apache License (Version 2.0). For models and datasets, please refer to the original resource page and follow the corresponding License.
🌐 Modelstudio Platform | 📖 Documentation | 🐛 Issues | 💬 Discussions
Built with ❤️ by the Alibaba Cloud Modelstudio Team
This project follows the all-contributors specification. Contributions of any kind welcome!